Add location recorder

src/util/location_recorder.py

@@ -0,0 +1,60 @@
+import datetime
+from dataclasses import dataclass
+
+from sqlalchemy import INTEGER, REAL, TEXT, insert
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class Version(Base):
+    __tablename__ = "version"
+    version_type: Mapped[str] = mapped_column(type_=TEXT, primary_key=True)
+    version: Mapped[int] = mapped_column(type_=INTEGER)
+
+
+class Location(Base):
+    __tablename__ = "location"
+    people: Mapped[str] = mapped_column(type_=TEXT, primary_key=True)
+    datetime: Mapped[str] = mapped_column(type_=TEXT, primary_key=True)
+    latitude: Mapped[float] = mapped_column(type_=REAL)
+    longitude: Mapped[float] = mapped_column(type_=REAL)
+    altitude: Mapped[float] = mapped_column(type_=REAL)
+
+
+@dataclass
+class LocationData:
+    latitude: float
+    longitude: float
+    altitude: float
+
+
+class LocationRecorder:
+    def __init__(self, db_path: str) -> None:
+        self._db_path = "sqlite+aiosqlite:///" + db_path
+
+    async def create_db_engine(self) -> None:
+        self._engine = create_async_engine(self._db_path)
+        async with self._engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+
+    async def insert_location(self, people: str, datetime: str, location: LocationData) -> None:
+        async with self._engine.connect() as conn:
+            await conn.execute(
+                insert(Location).values(
+                    people=people,
+                    datetime=datetime,
+                    latitude=location.latitude,
+                    longitude=location.longitude,
+                    altitude=location.altitude,
+                ),
+            )
+            await conn.commit()
+
+    async def insert_location_now(self, people: str, location: LocationData) -> None:
+        now = datetime.datetime.now(tz=datetime.UTC)
+        now_str = now.strftime("%Y-%m-%dT%H:%M:%S%z")
+        await self.insert_location(people, now_str, location)