Factor data.py out of strava_offline.py

This is a refactor in preparation of adding a CSV input.

src/strava_ical/data.py

@@ -0,0 +1,116 @@
+from datetime import datetime
+from datetime import timedelta
+from typing import Any
+from typing import Mapping
+from typing import Optional
+from typing import Tuple
+
+from dateutil.parser import isoparse
+
+essential_columns = {
+    'id',
+    'name',
+    'distance',
+    'total_elevation_gain',
+    'moving_time',
+    'elapsed_time',
+    'start_date',
+    'type',
+}
+
+
+class Activity:
+    def __init__(self, row: Mapping[str, Any]):
+        self._row = row
+
+    def __getitem__(self, key):
+        return self._row[key]
+
+    @property
+    def id(self) -> int:
+        return self['id']
+
+    @property
+    def name(self) -> str:
+        return self['name']
+
+    @property
+    def distance(self) -> float:
+        return self['distance']  # meters
+
+    @property
+    def total_elevation_gain(self) -> float:
+        return self['total_elevation_gain']  # meters
+
+    @property
+    def moving_time(self) -> timedelta:
+        return timedelta(seconds=self['moving_time'])
+
+    @property
+    def elapsed_time(self) -> timedelta:
+        return timedelta(seconds=self['elapsed_time'])
+
+    @property
+    def start_latlng(self) -> Optional[Tuple[float, float]]:
+        try:
+            [lat, lng] = self['start_latlng']
+            return lat, lng
+        except ValueError:
+            return None
+
+    @property
+    def start_datetime(self) -> datetime:
+        return isoparse(self['start_date'])
+
+    @property
+    def end_datetime(self) -> datetime:
+        return self.start_datetime + self.elapsed_time
+
+    @property
+    def type(self) -> str:
+        return self['type']
+
+    # see https://developers.strava.com/docs/reference/#api-models-ActivityType
+    _type_emojis = {
+        'AlpineSki': '⛷',
+        'BackcountrySki': '🎿',
+        'Canoeing': '🛶',
+        'Crossfit': '🤸',
+        'EBikeRide': '🛵',
+        'Elliptical': '🏃',
+        'Golf': '🏌',
+        'Handcycle': '🚴',
+        'Hike': '🥾',
+        'IceSkate': '⛸',
+        'InlineSkate': '🛼',
+        'Kayaking': '🛶',
+        'Kitesurf': '🏄',
+        'NordicSki': '🎿',
+        'Ride': '🚴',
+        'RockClimbing': '🧗',
+        'RollerSki': '🎿',
+        'Rowing': '🚣',
+        'Run': '🏃',
+        'Sail': '⛵',
+        'Skateboard': '🛹',
+        'Snowboard': '🏂',
+        'Snowshoe': '🎿',
+        'Soccer': '⚽',
+        'StairStepper': '🪜',
+        'StandUpPaddling': '🛶',
+        'Surfing': '🏄',
+        'Swim': '🏊',
+        'Velomobile': '🏎',
+        'VirtualRide': '🚴',
+        'VirtualRun': '🏃',
+        'Walk': '🚶',
+        'WeightTraining': '🏋',
+        'Wheelchair': '🧑‍🦽',
+        'Windsurf': '🏄',
+        'Workout': '💪',
+        'Yoga': '🧘',
+    }
+
+    @property
+    def type_emoji(self) -> str:
+        return self._type_emojis.get(self.type, "⏱")

src/strava_ical/ical.py

@@ -2,7 +2,7 @@
 
 import icalendar  # type: ignore [import]
 
-from .strava_offline import Activity
+from .data import Activity
 
 
 def ical(activities: Iterable[Activity]) -> bytes:

src/strava_ical/strava_offline.py

@@ -1,117 +1,11 @@
-from datetime import datetime
-from datetime import timedelta
 import json
 from os import PathLike
 import sqlite3
-from typing import Any
-from typing import Dict
 from typing import Iterable
-from typing import Optional
-from typing import Tuple
 from typing import Union
 
-from dateutil.parser import isoparse
-
-
-class Activity:
-    def __init__(self, row: Dict[str, Any]):
-        self._row = row
-        self._json = json.loads(row['json'])
-
-    def __getitem__(self, key):
-        if key in self._row:
-            return self._row[key]
-        else:
-            return self._json[key]
-
-    @property
-    def id(self) -> int:
-        return self['id']
-
-    @property
-    def name(self) -> str:
-        return self['name']
-
-    @property
-    def distance(self) -> float:
-        return self['distance']  # meters
-
-    @property
-    def total_elevation_gain(self) -> float:
-        return self['total_elevation_gain']  # meters
-
-    @property
-    def moving_time(self) -> timedelta:
-        return timedelta(seconds=self['moving_time'])
-
-    @property
-    def elapsed_time(self) -> timedelta:
-        return timedelta(seconds=self['elapsed_time'])
-
-    @property
-    def start_latlng(self) -> Optional[Tuple[float, float]]:
-        try:
-            [lat, lng] = self['start_latlng']
-            return lat, lng
-        except ValueError:
-            return None
-
-    @property
-    def start_datetime(self) -> datetime:
-        return isoparse(self['start_date'])
-
-    @property
-    def end_datetime(self) -> datetime:
-        return self.start_datetime + self.elapsed_time
-
-    @property
-    def type(self) -> str:
-        return self['type']
-
-    # see https://developers.strava.com/docs/reference/#api-models-ActivityType
-    _type_emojis = {
-        'AlpineSki': '⛷',
-        'BackcountrySki': '🎿',
-        'Canoeing': '🛶',
-        'Crossfit': '🤸',
-        'EBikeRide': '🛵',
-        'Elliptical': '🏃',
-        'Golf': '🏌',
-        'Handcycle': '🚴',
-        'Hike': '🥾',
-        'IceSkate': '⛸',
-        'InlineSkate': '🛼',
-        'Kayaking': '🛶',
-        'Kitesurf': '🏄',
-        'NordicSki': '🎿',
-        'Ride': '🚴',
-        'RockClimbing': '🧗',
-        'RollerSki': '🎿',
-        'Rowing': '🚣',
-        'Run': '🏃',
-        'Sail': '⛵',
-        'Skateboard': '🛹',
-        'Snowboard': '🏂',
-        'Snowshoe': '🎿',
-        'Soccer': '⚽',
-        'StairStepper': '🪜',
-        'StandUpPaddling': '🛶',
-        'Surfing': '🏄',
-        'Swim': '🏊',
-        'Velomobile': '🏎',
-        'VirtualRide': '🚴',
-        'VirtualRun': '🏃',
-        'Walk': '🚶',
-        'WeightTraining': '🏋',
-        'Wheelchair': '🧑‍🦽',
-        'Windsurf': '🏄',
-        'Workout': '💪',
-        'Yoga': '🧘',
-    }
-
-    @property
-    def type_emoji(self) -> str:
-        return self._type_emojis.get(self.type, "⏱")
+from .data import Activity
+from .data import essential_columns
 
 
 def read_strava_offline(db_filename: Union[str, PathLike]) -> Iterable[Activity]:
@@ -122,4 +16,7 @@ def read_strava_offline(db_filename: Union[str, PathLike]) -> Iterable[Activity]
         db.row_factory = sqlite3.Row
 
         for r in db.execute('SELECT * FROM activity'):
-            yield Activity({**r})
+            r_json = json.loads(r['json'])
+            r = {**r_json, **r}
+            assert essential_columns <= set(r.keys())
+            yield Activity(r)