Create and use new Location.from_json method to allow for reusable JSON parsing

Author: devinslick

fmd_api/device.py

@@ -44,22 +44,7 @@ async def refresh(self, *, force: bool = False):
 
         # decrypt and parse JSON
         decrypted = self.client.decrypt_data_blob(blobs[0])
-        loc = json.loads(decrypted)
-        # Build Location object with fields from README / fmd_api.py
-        timestamp_ms = loc.get("date")
-        ts = datetime.fromtimestamp(timestamp_ms / 1000.0, tz=timezone.utc) if timestamp_ms else None
-        self.cached_location = Location(
-            lat=loc["lat"],
-            lon=loc["lon"],
-            timestamp=ts,
-            accuracy_m=loc.get("accuracy"),
-            altitude_m=loc.get("altitude"),
-            speed_m_s=loc.get("speed"),
-            heading_deg=loc.get("heading"),
-            battery_pct=loc.get("bat"),
-            provider=loc.get("provider"),
-            raw=loc,
-        )
+        self.cached_location = Location.from_json(decrypted.decode("utf-8"))
 
     async def get_location(self, *, force: bool = False) -> Optional[Location]:
         if force or self.cached_location is None:
@@ -81,21 +66,7 @@ async def get_history(self, start=None, end=None, limit: int = -1) -> AsyncItera
         for b in blobs:
             try:
                 decrypted = self.client.decrypt_data_blob(b)
-                loc = json.loads(decrypted)
-                timestamp_ms = loc.get("date")
-                ts = datetime.fromtimestamp(timestamp_ms / 1000.0, tz=timezone.utc) if timestamp_ms else None
-                yield Location(
-                    lat=loc["lat"],
-                    lon=loc["lon"],
-                    timestamp=ts,
-                    accuracy_m=loc.get("accuracy"),
-                    altitude_m=loc.get("altitude"),
-                    speed_m_s=loc.get("speed"),
-                    heading_deg=loc.get("heading"),
-                    battery_pct=loc.get("bat"),
-                    provider=loc.get("provider"),
-                    raw=loc,
-                )
+                yield Location.from_json(decrypted.decode("utf-8"))
             except Exception as e:
                 # skip invalid blobs but log
                 raise OperationError(f"Failed to decrypt/parse location blob: {e}") from e

fmd_api/models.py

@@ -1,6 +1,7 @@
 from dataclasses import dataclass
-from datetime import datetime
-from typing import Optional, Dict, Any
+from datetime import datetime, timezone
+from typing import Optional, Dict, Any, Union
+import json as _json
 
 
 @dataclass
@@ -16,6 +17,51 @@ class Location:
     provider: Optional[str] = None
     raw: Optional[Dict[str, Any]] = None
 
+    @classmethod
+    def from_json(cls, json: Union[str, Dict[str, Any]]) -> "Location":
+        """Construct a Location from a JSON dict or JSON string.
+
+        Expected fields (from server payloads):
+        - lat (float)
+        - lon (float)
+        - date (int milliseconds since epoch)
+        - Optional: accuracy, altitude, speed, heading, bat, provider
+        """
+        # Accept either a JSON string or a dict
+        if isinstance(json, str):
+            try:
+                data = _json.loads(json)
+            except Exception as e:
+                raise ValueError(f"Invalid JSON string for Location: {e}") from e
+        elif isinstance(json, dict):
+            data = json
+        else:
+            raise TypeError("Location.from_json expects a dict or JSON string")
+
+        if "lat" not in data or "lon" not in data:
+            raise ValueError("Location JSON must include 'lat' and 'lon'")
+
+        # Convert date (ms since epoch) to aware datetime in UTC if present
+        ts = None
+        if data.get("date") is not None:
+            try:
+                ts = datetime.fromtimestamp(float(data["date"]) / 1000.0, tz=timezone.utc)
+            except Exception as e:
+                raise ValueError(f"Invalid 'date' field for Location: {e}") from e
+
+        return cls(
+            lat=float(data["lat"]),
+            lon=float(data["lon"]),
+            timestamp=ts,
+            accuracy_m=(float(data["accuracy"]) if data.get("accuracy") is not None else None),
+            altitude_m=(float(data["altitude"]) if data.get("altitude") is not None else None),
+            speed_m_s=(float(data["speed"]) if data.get("speed") is not None else None),
+            heading_deg=(float(data["heading"]) if data.get("heading") is not None else None),
+            battery_pct=(int(data["bat"]) if data.get("bat") is not None else None),
+            provider=(str(data["provider"]) if data.get("provider") is not None else None),
+            raw=data,
+        )
+
 
 @dataclass
 class PhotoResult:

tests/unit/test_models.py

@@ -0,0 +1,70 @@
+import json
+from datetime import timezone
+
+import pytest
+
+from fmd_api.models import Location
+
+
+def test_location_from_json_dict_basic():
+    data = {
+        "lat": 10.5,
+        "lon": 20.25,
+        "date": 1600000000000,  # ms since epoch
+        "accuracy": 5.0,
+        "altitude": 100.0,
+        "speed": 1.5,
+        "heading": 180.0,
+        "bat": 75,
+        "provider": "gps",
+    }
+    loc = Location.from_json(data)
+    assert loc.lat == 10.5
+    assert loc.lon == 20.25
+    assert loc.timestamp is not None
+    assert loc.timestamp.tzinfo == timezone.utc
+    assert loc.accuracy_m == 5.0
+    assert loc.altitude_m == 100.0
+    assert loc.speed_m_s == 1.5
+    assert loc.heading_deg == 180.0
+    assert loc.battery_pct == 75
+    assert loc.provider == "gps"
+    assert loc.raw == data
+
+
+def test_location_from_json_string_basic():
+    payload = {
+        "lat": 1.0,
+        "lon": 2.0,
+        "date": 1600000000000,
+    }
+    loc = Location.from_json(json.dumps(payload))
+    assert loc.lat == 1.0
+    assert loc.lon == 2.0
+    assert loc.timestamp is not None
+    assert loc.timestamp.tzinfo == timezone.utc
+
+
+def test_location_from_json_missing_optional_fields():
+    payload = {"lat": 0.0, "lon": 0.0, "date": 1600000000000}
+    loc = Location.from_json(payload)
+    assert loc.accuracy_m is None
+    assert loc.altitude_m is None
+    assert loc.speed_m_s is None
+    assert loc.heading_deg is None
+    assert loc.battery_pct is None
+    assert loc.provider is None
+
+
+def test_location_from_json_invalid_inputs():
+    with pytest.raises(TypeError):
+        Location.from_json(123)  # type: ignore[arg-type]
+
+    with pytest.raises(ValueError):
+        Location.from_json("not json")
+
+    with pytest.raises(ValueError):
+        Location.from_json({"lat": 1.0})  # missing lon
+
+    with pytest.raises(ValueError):
+        Location.from_json({"lat": 1.0, "lon": 2.0, "date": "abc"})