diff options
| -rw-r--r-- | fietsboek/updater/scripts/upd_20251109_nm561argcq1s8w27.py | 156 |
1 files changed, 156 insertions, 0 deletions
diff --git a/fietsboek/updater/scripts/upd_20251109_nm561argcq1s8w27.py b/fietsboek/updater/scripts/upd_20251109_nm561argcq1s8w27.py new file mode 100644 index 0000000..7c0fcae --- /dev/null +++ b/fietsboek/updater/scripts/upd_20251109_nm561argcq1s8w27.py @@ -0,0 +1,156 @@ +"""Revision upgrade script nm561argcq1s8w27 + +This script moves data from the GPX files in the data directory to the SQL +database. + +Date created: 2025-11-09 18:27:48.493007 +""" +import datetime +import logging +import re +import shutil +from pathlib import Path + +import brotli +import gpxpy +from sqlalchemy import create_engine +from sqlalchemy.sql import text + +from fietsboek import convert +from fietsboek.updater.script import UpdateScript + +LOGGER = logging.getLogger(__name__) + +update_id = 'nm561argcq1s8w27' +previous = [ + 'v0.11.0', +] +alembic_revision = '90b39fdf6e4b' + + +class Up(UpdateScript): + def pre_alembic(self, config): + pass + + def post_alembic(self, config): + engine = create_engine(config["sqlalchemy.url"]) + connection = engine.connect() + data_dir = Path(config["fietsboek.data_dir"]) + + for track_dir in (data_dir / "tracks").iterdir(): + track_id = int(track_dir.name) + self.tell(f"Loading track {track_id}") + + gpx_path = track_dir / "track.gpx.br" + + # We're careful here, in case a previous update was interrupted + if not gpx_path.exists(): + continue + + gpx_bytes = brotli.decompress(gpx_path.read_bytes()) + + track = convert.smart_convert(gpx_bytes) + with connection.begin(): + connection.execute( + text("DELETE FROM track_points WHERE track_id = :id;"), + {"id": track_id}, + ) + connection.execute( + text("DELETE FROM waypoints WHERE track_id = :id;"), + {"id": track_id}, + ) + for index, point in enumerate(track.path().points): + connection.execute( + text("""INSERT INTO track_points ( + track_id, "index", longitude, latitude, elevation, time_offset + ) VALUES ( + :track_id, :index, :longitude, :latitude, :elevation, :time_offset + );"""), + { + "track_id": track_id, + "index": index, + "longitude": point.longitude, + "latitude": point.latitude, + "elevation": point.elevation, + "time_offset": point.time_offset, + }, + ) + for waypoint in track.waypoints: + connection.execute( + text("""INSERT INTO waypoints ( + track_id, longitude, latitude, elevation, name, description + ) VALUES ( + :track_id, :longitude, :latitude, :elevation, :name, :description + );"""), + { + "track_id": track_id, + "longitude": waypoint.longitude, + "latitude": waypoint.latitude, + "elevation": waypoint.elevation, + "name": waypoint.name, + "description": waypoint.description, + }, + ) + + gpx_path.unlink() + shutil.move( + track_dir / "track.bck.gpx.br", + track_dir / "track.bck.br", + ) + +class Down(UpdateScript): + def pre_alembic(self, config): + engine = create_engine(config["sqlalchemy.url"]) + connection = engine.connect() + data_dir = Path(config["fietsboek.data_dir"]) + + query = text("SELECT id, title, description FROM tracks;") + + for row in connection.execute(query): + gpx = gpxpy.gpx.GPX() + gpx.description = row.description + gpx.name = row.title + + segment = gpxpy.gpx.GPXTrackSegment() + points_query = text(""" + SELECT longitude, latitude, elevation, time_offset + FROM track_points WHERE track_id = :track_id ORDER BY "index"; + """) + for point in connection.execute(points_query, {"track_id": row.id}): + segment.points.append( + gpxpy.gpx.GPXTrackPoint( + latitude=point.latitude, + longitude=point.longitude, + elevation=point.elevation, + # This is wrong, should take proper track start time + # into account: + time=datetime.datetime.fromtimestamp(point.time_offset), + ) + ) + track = gpxpy.gpx.GPXTrack() + track.segments.append(segment) + gpx.tracks.append(track) + + waypoints_query = text(""" + SELECT longitude, latitude, elevation, name, description + FROM waypoints WHERE track_id = :track_id; + """) + for wpt in connection.execute(waypoints_query, {"track_id": row.id}): + gpx.waypoints.append( + gpxpy.gpx.GPXWaypoint( + longitude=wpt.longitude, + latitude=wpt.latitude, + elevation=wpt.elevation, + name=wpt.name, + comment=wpt.description, + description=wpt.description, + ) + ) + + xml_data = gpx.to_xml(prettyprint=False).encode("utf-8") + track_dir = data_dir / "tracks" / str(row.id) + (track_dir / "track.gpx.br").write_bytes(brotli.compress(xml_data)) + shutil.move(track_dir / "track.bck.br", track_dir / "track.bck.gpx.br") + + def post_alembic(self, config): + pass |