Remove pydantic (#168)

Author: joostlek

poetry.lock

@@ -28,7 +28,6 @@ packages = [
 python = "^3.10"
 aiohttp = ">=3.0.0"
 yarl = ">=1.6.0"
-pydantic = ">=1.10.8"
 syrupy = "^4.5.0"
 
 [tool.poetry.group.dev.dependencies]

pyproject.toml

@@ -1,10 +1,14 @@
 """Asynchronous Python client for the OpenSky API."""
+import logging
 from enum import Enum
 
+LOGGER = logging.getLogger(__package__)
+
 
 class PositionSource(int, Enum):
     """Enum holding the Position source."""
 
+    UNKNOWN = -1
     ADSB = 0
     ASTERIX = 1
     MLAT = 2

src/python_opensky/const.py

@@ -2,21 +2,37 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-
-from pydantic import BaseModel, Field
+from typing import TYPE_CHECKING, Any
 
 from .const import AircraftCategory, PositionSource
 from .exceptions import OpenSkyCoordinateError
+from .util import to_enum
+
+if TYPE_CHECKING:
+    from typing_extensions import Self
 
 
-class StatesResponse(BaseModel):
+@dataclass(slots=True)
+class StatesResponse:
     """Represents the states response."""
 
-    states: list[StateVector] = Field(...)
-    time: int = Field(...)
+    states: list[StateVector]
+    time: int
+
+    @classmethod
+    def from_api(cls, data: dict[str, Any]) -> Self:
+        """Initialize from the API."""
+        return cls(
+            time=data["time"],
+            states=[
+                StateVector.from_api(vector_data) for vector_data in data["states"]
+            ],
+        )
 
 
-class StateVector(BaseModel):
+@dataclass(slots=True)
+# pylint: disable-next=too-many-instance-attributes
+class StateVector:
     """Represents the state of a vehicle at a particular time.
 
     Attributes
@@ -43,27 +59,59 @@ class StateVector(BaseModel):
     category: Aircraft category.
     """
 
-    icao24: str = Field(...)
-    callsign: str | None = Field(None)
-    origin_country: str = Field(...)
-    time_position: int | None = Field(None)
-    last_contact: int = Field(...)
-    longitude: float | None = Field(None)
-    latitude: float | None = Field(None)
-    geo_altitude: float | None = Field(None)
-    on_ground: bool = Field(...)
-    velocity: float | None = Field(None)
-    true_track: float | None = Field(None)
-    vertical_rate: float | None = Field(None)
-    sensors: list[int] | None = Field([])
-    barometric_altitude: float | None = Field(None, alias="baro_altitude")
-    transponder_code: str | None = Field(None, alias="squawk")
-    special_purpose_indicator: bool = Field(..., alias="spi")
-    position_source: PositionSource = Field(...)
-    category: AircraftCategory = Field(...)
-
-
-@dataclass
+    icao24: str
+    callsign: str | None
+    origin_country: str
+    time_position: int | None
+    last_contact: int
+    longitude: float | None
+    latitude: float | None
+    geo_altitude: float | None
+    on_ground: bool
+    velocity: float | None
+    true_track: float | None
+    vertical_rate: float | None
+    sensors: list[int] | None
+    barometric_altitude: float | None
+    transponder_code: str | None
+    special_purpose_indicator: bool
+    position_source: PositionSource
+    category: AircraftCategory
+
+    @classmethod
+    def from_api(cls, data: dict[str, Any]) -> Self:
+        """Initialize from the API."""
+        return cls(
+            icao24=data["icao24"],
+            callsign=data.get("callsign"),
+            origin_country=data["origin_country"],
+            time_position=data.get("time_position"),
+            last_contact=data["last_contact"],
+            longitude=data.get("longitude"),
+            latitude=data.get("latitude"),
+            geo_altitude=data.get("geo_altitude"),
+            on_ground=data["on_ground"],
+            velocity=data.get("velocity"),
+            true_track=data.get("true_track"),
+            vertical_rate=data.get("vertical_rate"),
+            sensors=data.get("sensors", []),
+            barometric_altitude=data.get("baro_altitude"),
+            transponder_code=data.get("squawk"),
+            special_purpose_indicator=data["spi"],
+            position_source=to_enum(
+                PositionSource,
+                data["position_source"],
+                PositionSource.UNKNOWN,
+            ),
+            category=to_enum(
+                AircraftCategory,
+                data["category"],
+                AircraftCategory.NO_INFORMATION,
+            ),
+        )
+
+
+@dataclass(slots=True)
 class BoundingBox:
     """Bounding box for retrieving state vectors."""
 
@@ -90,7 +138,3 @@ def _check_longitude(degrees: float) -> None:
         if degrees < -180 or degrees > 180:
             msg = f"Invalid longitude {degrees}! Must be in [-180, 180]."
             raise OpenSkyCoordinateError(msg)
-
-
-StatesResponse.update_forward_refs()
-StateVector.update_forward_refs()

src/python_opensky/models.py

@@ -178,7 +178,7 @@ async def get_states(
 
         self._register_credit_usage(credit_cost)
 
-        return StatesResponse.parse_obj(data)
+        return StatesResponse.from_api(data)
 
     async def get_own_states(self, time: int = 0) -> StatesResponse:
         """Retrieve state vectors from your own sensors."""
@@ -198,7 +198,7 @@ async def get_own_states(self, time: int = 0) -> StatesResponse:
                 "states": [self._convert_state(state) for state in data["states"]],
             }
 
-        return StatesResponse.parse_obj(data)
+        return StatesResponse.from_api(data)
 
     @staticmethod
     def calculate_credit_costs(bounding_box: BoundingBox) -> int:

src/python_opensky/opensky.py

@@ -0,0 +1,25 @@
+"""Asynchronous Python client for OpenSky."""
+from __future__ import annotations
+
+from typing import Any, TypeVar
+
+from .const import LOGGER
+
+_EnumT = TypeVar("_EnumT")
+
+
+def to_enum(
+    enum_class: type[_EnumT],
+    value: Any,
+    default_value: _EnumT,
+) -> _EnumT:
+    """Convert a value to an enum and log if it doesn't exist."""
+    try:
+        return enum_class(value)  # type: ignore[call-arg]
+    except ValueError:
+        LOGGER.warning(
+            "%s is an unsupported value for %s, please report this at https://github.com/joostlek/python-opensky/issues",
+            value,
+            str(enum_class),
+        )
+        return default_value

src/python_opensky/util.py

@@ -20,7 +20,7 @@
         'transponder_code': None,
         'true_track': 342.17,
         'velocity': 137.8,
-        'vertical_rate': 13.0,
+        'vertical_rate': 13,
       }),
       dict({
         'barometric_altitude': None,
@@ -65,7 +65,7 @@
       dict({
         'barometric_altitude': 701.04,
         'callsign': 'N28CN   ',
-        'category': <AircraftCategory.GLIDER: 9>,
+        'category': <AircraftCategory.NO_INFORMATION: 0>,
         'geo_altitude': 685.8,
         'icao24': 'a2cbe1',
         'last_contact': 1683488743,

tests/__snapshots__/test_states.ambr

@@ -3,7 +3,7 @@
     ["ab1644", "UAL421  ", "United States", 1683488743, 1683488743, -71.1656, 42.5372, 2217.42, false, 137.8, 342.17, 13, null, 2194.56, null, false, 0, 4],
     ["e8027e", "LPE2264 ", "Chile", 1683488479, 1683488722, -77.1296, -12.0115, null, true, 83.29, 149.16, 9.1, null, null, null, false, 0, 1],
     ["e8027d", "LPE2392 ", "Chile", 1683488743, 1683488743, -76.9275, -11.905, 4335.78, false, 187.32, 332.17, 10.4, null, null, null, false, 0, 1],
-    ["a2cbe1", "N28CN   ", "United States", 1683488743, 1683488743, -93.9953, 44.9762, 701.04, false, 73.29, 293.15, -4.55, null, 685.8, null, false, 0, 9]
+    ["a2cbe1", "N28CN   ", "United States", 1683488743, 1683488743, -93.9953, 44.9762, 701.04, false, 73.29, 293.15, -4.55, null, 685.8, null, false, 0, -1]
   ],
   "time": 1683488744
 }

tests/fixtures/states.json

@@ -1,5 +1,6 @@
 """Tests for the OpenSky Library."""
 import asyncio
+from dataclasses import asdict
 
 import aiohttp
 import pytest
@@ -40,7 +41,7 @@ async def test_states(
     async with aiohttp.ClientSession() as session:
         opensky = OpenSky(session=session)
         response: StatesResponse = await opensky.get_states()
-        assert response.model_dump() == snapshot
+        assert asdict(response) == snapshot
         await opensky.close()