Add method to check latest version (#174)

src/airgradient/airgradient.py

@@ -21,6 +21,7 @@
     Measures,
     PmStandard,
     TemperatureUnit,
+    VersionCheck,
 )
 
 if TYPE_CHECKING:
@@ -41,14 +42,12 @@ class AirGradientClient:
 
     async def _request(
         self,
-        uri: str,
+        url: URL,
         *,
         method: str = METH_GET,
         data: dict[str, Any] | None = None,
     ) -> str:
         """Handle a request to AirGradient."""
-        url = URL.build(scheme="http", host=self.host).joinpath(uri)
-
         headers = {
             "User-Agent": f"PythonAirGradient/{VERSION}",
             "Accept": "application/json",
@@ -88,25 +87,32 @@ async def _request(
 
         return await response.text()
 
+    async def _request_device(
+        self, uri: str, *, method: str = METH_GET, data: dict[str, Any] | None = None
+    ) -> str:
+        """Handle a request to the AirGradient device."""
+        url = URL.build(scheme="http", host=self.host).joinpath(uri)
+        return await self._request(url, method=method, data=data)
+
     async def get_current_measures(self) -> Measures:
         """Get current measures from AirGradient."""
-        response = await self._request("measures/current")
+        response = await self._request_device("measures/current")
         try:
             return Measures.from_json(response)
         except MissingField as err:
             raise AirGradientParseError from err
 
     async def get_config(self) -> Config:
         """Get config from AirGradient device."""
-        response = await self._request("config")
+        response = await self._request_device("config")
         try:
             return Config.from_json(response)
         except MissingField as err:
             raise AirGradientParseError from err
 
     async def _set_config(self, field: str, value: Any) -> None:
         """Set config on AirGradient device."""
-        await self._request("config", method=METH_PUT, data={field: value})
+        await self._request_device("config", method=METH_PUT, data={field: value})
 
     async def set_pm_standard(self, pm_standard: PmStandard) -> None:
         """Set PM standard on AirGradient device."""
@@ -158,6 +164,17 @@ async def set_tvoc_learning_offset(self, offset: int) -> None:
         """Set TVOC learning offset on AirGradient device."""
         await self._set_config("tvocLearningOffset", offset)
 
+    async def get_latest_firmware_version(self, serial_number: str) -> str:
+        """Get the latest firmware version from AirGradient."""
+        url = URL.build(scheme="http", host="hw.airgradient.com").joinpath(
+            f"sensors/airgradient:{serial_number}/generic/os/firmware"
+        )
+        response = await self._request(url)
+        try:
+            return VersionCheck.from_json(response).target_version
+        except MissingField as err:
+            raise AirGradientParseError from err
+
     async def close(self) -> None:
         """Close open client session."""
         if self.session and self._close_session:

src/airgradient/models.py

@@ -100,3 +100,10 @@ class Config(DataClassORJSONMixin):
     tvoc_learning_offset: int = field(
         metadata=field_options(alias="tvocLearningOffset")
     )
+
+
+@dataclass
+class VersionCheck(DataClassORJSONMixin):
+    """Version check model."""
+
+    target_version: str = field(metadata=field_options(alias="targetVersion"))

tests/__snapshots__/test_airgradient.ambr

@@ -54,3 +54,6 @@
     'total_volatile_organic_component_index': None,
   })
 # ---
+# name: test_latest_version
+  '3.1.4'
+# ---

tests/fixtures/version.json

@@ -0,0 +1 @@
+{"targetVersion":"3.1.4"}

tests/test_airgradient.py

@@ -247,3 +247,34 @@ async def test_setting_config(
         headers=HEADERS,
         json=expected_data,
     )
+
+
+async def test_latest_version(
+    responses: aioresponses, client: AirGradientClient, snapshot: SnapshotAssertion
+) -> None:
+    """Test getting latest firmware version."""
+    responses.get(
+        "http://hw.airgradient.com/sensors/airgradient:84fce612f5b8/generic/os/firmware",
+        status=200,
+        body=load_fixture("version.json"),
+    )
+    assert snapshot == await client.get_latest_firmware_version("84fce612f5b8")
+    responses.assert_called_with(
+        "http://hw.airgradient.com/sensors/airgradient:84fce612f5b8/generic/os/firmware",
+        headers=HEADERS,
+        json=None,
+    )
+
+
+async def test_version_parse_error(
+    responses: aioresponses,
+    client: AirGradientClient,
+) -> None:
+    """Test version parse error."""
+    responses.get(
+        "http://hw.airgradient.com/sensors/airgradient:84fce612f5b8/generic/os/firmware",
+        status=200,
+        body="{}",
+    )
+    with pytest.raises(AirGradientParseError):
+        await client.get_latest_firmware_version("84fce612f5b8")