diff --git a/src/helper/location_recorder/google_location_reader.py b/src/helper/location_recorder/google_location_reader.py new file mode 100644 index 0000000..befbe17 --- /dev/null +++ b/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()) diff --git a/src/helper/location_recorder/gpx_location_reader.py b/src/helper/location_recorder/gpx_location_reader.py new file mode 100644 index 0000000..e8f0277 --- /dev/null +++ b/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()) diff --git a/src/util/location_recorder.py b/src/util/location_recorder.py index 8b82791..940d34c 100644 --- a/src/util/location_recorder.py +++ b/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) diff --git a/src/util/tests/test_location_recorder.py b/src/util/tests/test_location_recorder.py index b0af54b..71d78b5 100644 --- a/src/util/tests/test_location_recorder.py +++ b/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()