Add insert locations

src/helper/location_recorder/google_location_reader.py

@@ -0,0 +1,61 @@
+import argparse
+import asyncio
+import json
+import sys
+from datetime import datetime
+from pathlib import Path
+
+current_file_path = Path(__file__).resolve().parent
+sys.path.append(str(current_file_path / ".." / ".." / ".."))
+from src.util.location_recorder import LocationData, LocationRecorder  # noqa: E402
+
+# Create an argument parser
+parser = argparse.ArgumentParser(description="Google Location Reader")
+
+# Add an argument for the JSON file path
+parser.add_argument("--json-file", type=str, help="Path to the JSON file")
+
+# Parse the command-line arguments
+args = parser.parse_args()
+
+json_file_path: str = args.json_file
+
+db_path = current_file_path / ".." / ".." / ".." / "temp_data" / "test.db"
+location_recorder = LocationRecorder(db_path=str(db_path))
+
+# Open the JSON file
+with Path.open(json_file_path) as json_file:
+    data = json.load(json_file)
+
+
+locations: list[dict] = data["locations"]
+print(type(locations), len(locations))
+
+
+async def insert() -> None:
+    nr_waypoints = 0
+    await location_recorder.create_db_engine()
+    locations_dict: dict[datetime, LocationData] = {}
+    for location in locations:
+        nr_waypoints += 1
+        try:
+            latitude: float = location["latitudeE7"] / 1e7
+            longitude: float = location["longitudeE7"] / 1e7
+        except KeyError:
+            continue
+        altitude: float = location.get("altitude", None)
+        try:
+            date_time = datetime.strptime(location["timestamp"], "%Y-%m-%dT%H:%M:%S.%f%z")
+        except ValueError:
+            date_time = datetime.strptime(location["timestamp"], "%Y-%m-%dT%H:%M:%S%z")
+        locations_dict[date_time] = LocationData(
+            latitude=latitude,
+            longitude=longitude,
+            altitude=altitude,
+        )
+    await location_recorder.insert_locations("Tianyu", locations=locations_dict)
+    print(nr_waypoints)
+    await location_recorder.dispose_db_engine()
+
+
+asyncio.run(insert())

src/helper/location_recorder/gpx_location_reader.py

@@ -0,0 +1,44 @@
+import argparse
+import asyncio
+import sys
+from datetime import UTC
+from pathlib import Path
+
+import gpxpy
+import gpxpy.gpx
+
+current_file_path = Path(__file__).resolve().parent
+sys.path.append(str(current_file_path / ".." / ".." / ".."))
+from src.util.location_recorder import LocationData, LocationRecorder  # noqa: E402
+
+parser = argparse.ArgumentParser(description="GPX Location Reader")
+
+parser.add_argument("--gpx-file", type=str, help="Path to the GPX file")
+
+args = parser.parse_args()
+
+gpx_location = args.gpx_file
+
+gpx_file = Path.open(gpx_location)
+gpx = gpxpy.parse(gpx_file)
+
+db_path = current_file_path / ".." / ".." / ".." / "temp_data" / "test.db"
+location_recorder = LocationRecorder(db_path=str(db_path))
+
+
+async def iterate_and_insert() -> None:
+    nr_waypoints = 0
+    await location_recorder.create_db_engine()
+    for track in gpx.tracks:
+        for segment in track.segments:
+            for point in segment.points:
+                nr_waypoints += 1
+                print(f"Point at ({point.latitude},{point.longitude}) -> {point.time}")
+                point.time = point.time.replace(tzinfo=UTC)
+                location_data = LocationData(latitude=point.latitude, longitude=point.longitude, altitude=point.elevation)
+                await location_recorder.insert_location(person="Tianyu", date_time=point.time, location=location_data)
+    await location_recorder.dispose_db_engine()
+    print(nr_waypoints)
+
+
+asyncio.run(iterate_and_insert())

src/util/location_recorder.py

@@ -62,6 +62,25 @@ class LocationRecorder:
                 ),
             )
 
+    async def insert_locations(self, person: str, locations: dict[datetime, LocationData]) -> None:
+        async with self._engine.begin() as conn:
+            for k, v in locations.items():
+                dt = k
+                if k.tzinfo != UTC:
+                    dt = k.astimezone(UTC)
+                date_time_str = dt.strftime("%Y-%m-%dT%H:%M:%S%z")
+                await conn.execute(
+                    insert(Location)
+                    .prefix_with("OR IGNORE")
+                    .values(
+                        person=person,
+                        datetime=date_time_str,
+                        latitude=v.latitude,
+                        longitude=v.longitude,
+                        altitude=v.altitude,
+                    ),
+                )
+
     async def insert_location_now(self, person: str, location: LocationData) -> None:
         now_utc = datetime.now(tz=UTC)
         await self.insert_location(person, now_utc, location)

src/util/tests/test_location_recorder.py

@@ -312,3 +312,43 @@ def test_insert_location_now() -> None:
     assert location[3] == longitude
     assert location[4] == altitude
     sqlite3_cursor.close()
+
+
+@pytest.mark.usefixtures("_reset_event_loop")
+@pytest.mark.usefixtures("_create_latest_db")
+@pytest.mark.usefixtures("_teardown")
+def test_insert_locations() -> None:
+    locations: dict[datetime, LocationData] = {}
+    person = "Tianyu"
+    time_0 = datetime.now(tz=UTC)
+    lat_0 = 1.0
+    lon_0 = 2.0
+    alt_0 = 3.0
+    time_1 = datetime(2021, 8, 30, 10, 20, 15, tzinfo=UTC)
+    lat_1 = 155.0
+    lon_1 = 33.36
+    alt_1 = 1058
+    locations[time_0] = LocationData(lat_0, lon_0, alt_0)
+    locations[time_1] = LocationData(lat_1, lon_1, alt_1)
+    location_recorder = LocationRecorder(db_path=DB_PATH_STR)
+    event_loop = asyncio.get_event_loop()
+    event_loop.run_until_complete(location_recorder.create_db_engine())
+    event_loop.run_until_complete(
+        location_recorder.insert_locations(person=person, locations=locations),
+    )
+    sqlite3_db = sqlite3.connect(DB_PATH_STR)
+    sqlite3_cursor = sqlite3_db.cursor()
+    sqlite3_cursor.execute("SELECT * FROM location")
+    locations = sqlite3_cursor.fetchall()
+    assert len(locations) == 2  # noqa: PLR2004
+    assert locations[0][0] == person
+    assert locations[0][1] == time_0.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%S%z")
+    assert locations[0][2] == lat_0
+    assert locations[0][3] == lon_0
+    assert locations[0][4] == alt_0
+    assert locations[1][0] == person
+    assert locations[1][1] == time_1.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%S%z")
+    assert locations[1][2] == lat_1
+    assert locations[1][3] == lon_1
+    assert locations[1][4] == alt_1
+    sqlite3_cursor.close()