Restruct files, prepare for next feature

src/config.py

@@ -0,0 +1,7 @@
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+NOTION_TOKEN = os.getenv("NOTION_TOKEN")

src/main.py

@@ -0,0 +1,23 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from src.recorder.poo import PooRecorder
+
+recorder = PooRecorder()
+
+
+@asynccontextmanager
+async def _lifespan(_app: FastAPI):  # noqa: ANN202
+    await recorder.start()
+    yield
+    await recorder.stop()
+
+
+app = FastAPI(lifespan=_lifespan)
+
+
+@app.put("/record/s={status}")
+async def record(status: str) -> dict:
+    await recorder.record(status)
+    return {"status": status}

src/recorder/notion_handle.py

@@ -0,0 +1,32 @@
+from datetime import datetime
+
+from notion_client import AsyncClient as Client
+
+from src.config import NOTION_TOKEN
+
+
+class NotionClient:
+    def __init__(self) -> None:
+        self._notion = Client(auth=NOTION_TOKEN)
+        self._page_id = "3cf594afd0754497ba0a93b94912b897"
+        self._table_id = "9828b56c53de46c794673fe1d01ad522"
+
+    async def note(self, now: datetime, status: str) -> None:
+        formatted_date = now.strftime("%Y-%m-%d")
+        formatted_time = now.strftime("%H:%M")
+        await self._notion.blocks.children.append(
+            block_id=self._table_id,
+            children=[
+                {
+                    "object": "block",
+                    "type": "table_row",
+                    "table_row": {
+                        "cells": [
+                            [{"type": "text", "text": {"content": formatted_date}}],
+                            [{"type": "text", "text": {"content": formatted_time}}],
+                            [{"type": "text", "text": {"content": status}}],
+                        ],
+                    },
+                },
+            ],
+        )

src/recorder/poo.py

@@ -0,0 +1,77 @@
+from datetime import datetime
+
+from fastapi_mqtt import FastMQTT, MQTTConfig
+
+from src.recorder.notion_handle import NotionClient
+
+
+class PooRecorder:
+    mqtt_config = MQTTConfig(username="mqtt", password="mqtt", reconnect_retries=-1)  # noqa: S106
+    notion = NotionClient()
+    mqtt = FastMQTT(config=mqtt_config, client_id="poo_recorder")
+    CONFIG_TOPIC = "homeassistant/text/poo_recorder/config"
+    AVAILABILITY_TOPIC = "studiotj/poo_recorder/status"
+    COMMAND_TOPIC = "studiotj/poo_recorder/update_text"
+    STATE_TOPIC = "studiotj/poo_recorder/text"
+    JSON_TOPIC = "studiotj/poo_recorder/attributes"
+    ONLINE = "online"
+    OFFLINE = "offline"
+
+    def __init__(self) -> None:
+        print("Initialization.")
+
+    async def start(self) -> None:
+        print("Starting...")
+        await PooRecorder.mqtt.mqtt_startup()
+
+    async def stop(self) -> None:
+        print("Stopping...")
+        await PooRecorder.mqtt.mqtt_shutdown()
+
+    async def record(self, status: str) -> None:
+        PooRecorder.publish_text(status)
+        now = datetime.now(tz=datetime.now().astimezone().tzinfo)
+        PooRecorder.publish_time(now)
+        await PooRecorder.notion.note(now, status)
+
+    @staticmethod
+    @mqtt.on_connect()
+    def on_connect(client, flags, rc, properties) -> None:  # noqa: ANN001, ARG004
+        print("Connected")
+        config = PooRecorder.compose_config()
+        PooRecorder.mqtt.publish(PooRecorder.CONFIG_TOPIC, config, retain=True)
+        PooRecorder.mqtt.publish(PooRecorder.AVAILABILITY_TOPIC, PooRecorder.ONLINE, retain=True)
+
+    @staticmethod
+    def publish_text(new_text: str) -> None:
+        PooRecorder.mqtt.publish(PooRecorder.AVAILABILITY_TOPIC, PooRecorder.ONLINE, retain=True)
+        PooRecorder.mqtt.publish(PooRecorder.STATE_TOPIC, new_text, retain=True)
+
+    @staticmethod
+    def publish_time(time: datetime) -> None:
+        formatted_time = time.strftime("%a | %Y-%m-%d | %H:%M")
+        PooRecorder.mqtt.publish(PooRecorder.AVAILABILITY_TOPIC, PooRecorder.ONLINE, retain=True)
+        json_string = {"last_poo": formatted_time}
+        PooRecorder.mqtt.publish(PooRecorder.JSON_TOPIC, json_string, retain=True)
+
+    @staticmethod
+    def compose_config() -> dict:
+        return {
+            "device": {
+                "name": "Dog Poop Recorder",
+                "model": "poop-recorder-backend",
+                "sw_version": "1.1",
+                "identifiers": ["poo_recorder"],
+                "manufacturer": "Studio TJ",
+            },
+            "unique_id": "poo_recorder",
+            "name": "Poo Status",
+            "availability_topic": PooRecorder.AVAILABILITY_TOPIC,
+            "availability_template": "{{ value_json.availability }}",
+            "json_attributes_topic": PooRecorder.JSON_TOPIC,
+            "min": 0,
+            "max": 255,
+            "mode": "text",
+            "command_topic": PooRecorder.COMMAND_TOPIC,
+            "state_topic": PooRecorder.STATE_TOPIC,
+        }