Add endpoint for GET /sensor/:sensor_index

README.md

@@ -18,6 +18,7 @@
 - [Usage](#usage)
   - [Checking an API Key](#checking-an-api-key)
   - [Getting Sensors](#getting-sensors)
+  - [Getting a Single Sensor](#getting-a-single-sensor)
   - [Connection Pooling](#connection-pooling)
 - [Contributing](#contributing)
 
@@ -101,12 +102,42 @@ asyncio.run(main())
 
 `API.sensors.async_get_sensors` takes several parameters:
 
-- `fields`: The sensor data fields to include.
-- `location_type`: An optional LocationType to filter by.
-- `max_age`: Filter results modified within these seconds.
-- `modified_since`: Filter results modified since a UTC datetime.
-- `read_keys`: Optional read keys for private sensors.
-- `sensor_indices`: Filter results by sensor index.
+- `fields` (required): The sensor data fields to include.
+- `location_type` (optional): An LocationType to filter by.
+- `max_age` (optional): Filter results modified within these seconds.
+- `modified_since` (optional): Filter results modified since a UTC datetime.
+- `read_keys` (optional): Read keys for private sensors.
+- `sensor_indices` (optional): Filter results by sensor index.
+
+## Getting a Single Sensor
+
+```python
+import asyncio
+
+from aiopurpleair import API
+
+
+async def main() -> None:
+    """Run."""
+    api = API("<API_KEY>")
+    response = await api.sensors.async_get_sensor("<SENSOR INDEX>")
+    # >>> response.api_version == "V1.0.11-0.0.41"
+    # >>> response.time_stamp == datetime(2022, 11, 5, 16, 37, 3)
+    # >>> response.data_time_stamp == datetime(2022, 11, 5, 16, 36, 21)
+    # >>> response.sensor == {
+    # >>>     "sensor_index": 131075,
+    # >>>     "last_modified": 1635632829,
+    # >>>     ...
+    # >>> }
+
+
+asyncio.run(main())
+```
+
+`API.sensors.async_get_sensor` takes several parameters:
+
+- `fields` (optional): The sensor data fields to include.
+- `read_key` (optional): A read key for a private sensor.
 
 ## Connection Pooling
 

aiopurpleair/api.py

@@ -10,7 +10,7 @@
 from aiopurpleair.const import LOGGER
 from aiopurpleair.endpoints.sensors import SensorsEndpoints
 from aiopurpleair.errors import RequestError, raise_error
-from aiopurpleair.helpers.typing import ResponseModelT
+from aiopurpleair.helpers.typing import ModelT
 from aiopurpleair.models.keys import GetKeysResponse
 
 API_URL_BASE = "https://api.purpleair.com/v1"
@@ -60,7 +60,7 @@ async def async_request(
         endpoint: str,
         response_model: type[BaseModel],
         **kwargs: dict[str, Any],
-    ) -> ResponseModelT:
+    ) -> ModelT:
         """Make an API request.
 
         Args:
@@ -106,7 +106,7 @@ async def async_request(
         LOGGER.debug("Data received for %s: %s", endpoint, data)
 
         try:
-            return cast(ResponseModelT, response_model.parse_obj(data))
+            return cast(ModelT, response_model.parse_obj(data))
         except ValidationError as err:
             raise RequestError(
                 f"Error while parsing response from {endpoint}: {err}"

aiopurpleair/const.py

@@ -2,3 +2,130 @@
 import logging
 
 LOGGER = logging.getLogger(__package__)
+
+SENSOR_FIELDS = {
+    "0.3_um_count",
+    "0.3_um_count_a",
+    "0.3_um_count_b",
+    "0.5_um_count",
+    "0.5_um_count_a",
+    "0.5_um_count_b",
+    "1.0_um_count",
+    "1.0_um_count_a",
+    "1.0_um_count_b",
+    "10.0_um_count 10.0_um_count_a",
+    "10.0_um_count_b",
+    "2.5_um_count",
+    "2.5_um_count_a",
+    "2.5_um_count_b",
+    "5.0_um_count",
+    "5.0_um_count_a",
+    "5.0_um_count_b",
+    "altitude",
+    "analog_input",
+    "channel_flags",
+    "channel_flags_auto",
+    "channel_flags_manual",
+    "channel_state",
+    "confidence",
+    "confidence_auto",
+    "confidence_manual",
+    "date_created",
+    "deciviews",
+    "deciviews_a",
+    "deciviews_b",
+    "firmware_upgrade",
+    "firmware_version",
+    "hardware",
+    "humidity",
+    "humidity_a",
+    "humidity_b",
+    "icon",
+    "last_modified",
+    "last_seen",
+    "latitude",
+    "led_brightness",
+    "location_type",
+    "longitude",
+    "memory",
+    "model",
+    "name",
+    "ozone1",
+    "pa_latency",
+    "pm1.0",
+    "pm1.0_a",
+    "pm1.0_atm",
+    "pm1.0_atm_a",
+    "pm1.0_atm_b",
+    "pm1.0_b",
+    "pm1.0_cf_1",
+    "pm1.0_cf_1_a",
+    "pm1.0_cf_1_b",
+    "pm10.0",
+    "pm10.0_a",
+    "pm10.0_atm",
+    "pm10.0_atm_a",
+    "pm10.0_atm_b",
+    "pm10.0_b",
+    "pm10.0_cf_1",
+    "pm10.0_cf_1_a",
+    "pm10.0_cf_1_b",
+    "pm2.5",
+    "pm2.5_10minute",
+    "pm2.5_10minute_a",
+    "pm2.5_10minute_b",
+    "pm2.5_1week",
+    "pm2.5_1week_a",
+    "pm2.5_1week_b",
+    "pm2.5_24hour",
+    "pm2.5_24hour_a",
+    "pm2.5_24hour_b",
+    "pm2.5_30minute",
+    "pm2.5_30minute_a",
+    "pm2.5_30minute_b",
+    "pm2.5_60minute",
+    "pm2.5_60minute_a",
+    "pm2.5_60minute_b",
+    "pm2.5_6hour",
+    "pm2.5_6hour_a",
+    "pm2.5_6hour_b",
+    "pm2.5_a",
+    "pm2.5_alt",
+    "pm2.5_alt_a",
+    "pm2.5_alt_b",
+    "pm2.5_atm",
+    "pm2.5_atm_a",
+    "pm2.5_atm_b",
+    "pm2.5_b",
+    "pm2.5_cf_1",
+    "pm2.5_cf_1_a",
+    "pm2.5_cf_1_b",
+    "position_rating",
+    "pressure",
+    "pressure_a",
+    "pressure_b",
+    "primary_id_a",
+    "primary_id_b",
+    "primary_key_a",
+    "primary_key_b",
+    "private",
+    "rssi",
+    "scattering_coefficient",
+    "scattering_coefficient_a",
+    "scattering_coefficient_b",
+    "secondary_id_a",
+    "secondary_id_b",
+    "secondary_key_a",
+    "secondary_key_b",
+    "sensor_index",
+    "temperature",
+    "temperature_a",
+    "temperature_b",
+    "uptime",
+    "visual_range",
+    "visual_range_a",
+    "visual_range_b",
+    "voc",
+    "voc_a",
+    "voc_b",
+}

aiopurpleair/endpoints/__init__.py

@@ -1 +1,59 @@
 """Define API endpoints."""
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Iterable
+from typing import Any
+
+from pydantic import BaseModel, ValidationError
+
+from aiopurpleair.errors import InvalidRequestError
+from aiopurpleair.helpers.typing import ModelT
+
+
+class APIEndpointsBase:  # pylint: disable=too-few-public-methods
+    """Define a base API endpoints manager."""
+
+    def __init__(self, async_request: Callable[..., Awaitable[ModelT]]) -> None:
+        """Initialize.
+
+        Args:
+            async_request: The request method from the API object.
+        """
+        self._async_request = async_request
+
+    async def _async_endpoint_request(
+        self,
+        endpoint: str,
+        query_param_map: Iterable[tuple[str, Any]],
+        request_model: type[BaseModel],
+        response_model: type[BaseModel],
+    ) -> ModelT:
+        """Perform an API endpoint request.
+
+        Args:
+            endpoint: The API endpoint to query.
+            query_param_map: A tuple of API query parameters to include (if they exist).
+            request_model: The Pydantic model for the request.
+            response_model: The Pydantic model for the response.
+
+        Returns:
+            An API response payload in the form of a Pydantic model.
+
+        Raises:
+            InvalidRequestError: Raised on invalid parameters.
+        """
+        payload: dict[str, Any] = {}
+
+        for api_query_param, func_param in query_param_map:
+            if not func_param:
+                continue
+            payload[api_query_param] = func_param
+
+        try:
+            request = request_model.parse_obj(payload)
+        except ValidationError as err:
+            raise InvalidRequestError(err) from err
+
+        return await self._async_request(
+            "get", endpoint, response_model, params=request.dict(exclude_none=True)
+        )

aiopurpleair/endpoints/sensors.py

@@ -1,31 +1,48 @@
 """Define an API endpoint for requests related to sensors."""
 from __future__ import annotations
 
-from collections.abc import Awaitable, Callable
 from datetime import datetime
-from typing import Any, cast
 
-from pydantic import ValidationError
-
-from aiopurpleair.errors import InvalidRequestError
-from aiopurpleair.helpers.typing import ResponseModelT
+from aiopurpleair.endpoints import APIEndpointsBase
 from aiopurpleair.models.sensors import (
+    GetSensorRequest,
+    GetSensorResponse,
     GetSensorsRequest,
     GetSensorsResponse,
     LocationType,
 )
 
 
-class SensorsEndpoints:  # pylint: disable=too-few-public-methods
+class SensorsEndpoints(APIEndpointsBase):
     """Define the API manager object."""
 
-    def __init__(self, async_request: Callable[..., Awaitable[ResponseModelT]]) -> None:
-        """Initialize.
+    async def async_get_sensor(
+        self,
+        sensor_index: int,
+        *,
+        fields: list[str] | None = None,
+        read_key: str | None = None,
+    ) -> GetSensorResponse:
+        """Get all sensors.
 
         Args:
-            async_request: The request method from the API object.
+            sensor_index: The sensor index to get data for.
+            fields: The optional sensor data fields to include.
+            read_key: An optional read key for private sensors.
+
+        Returns:
+            An API response payload in the form of a Pydantic model.
         """
-        self._async_request = async_request
+        response: GetSensorResponse = await self._async_endpoint_request(
+            f"/sensor/{sensor_index}",
+            (
+                ("fields", fields),
+                ("read_key", read_key),
+            ),
+            GetSensorRequest,
+            GetSensorResponse,
+        )
+        return response
 
     async def async_get_sensors(
         self,
@@ -49,32 +66,18 @@ async def async_get_sensors(
 
         Returns:
             An API response payload in the form of a Pydantic model.
-
-        Raises:
-            InvalidRequestError: Raised on invalid parameters.
         """
-        payload: dict[str, Any] = {"fields": fields}
-
-        for api_query_param, func_param in (
-            ("location_type", location_type),
-            ("max_age", max_age),
-            ("modified_since", modified_since),
-            ("read_keys", read_keys),
-            ("show_only", sensor_indices),
-        ):
-            if not func_param:
-                continue
-            payload[api_query_param] = func_param
-
-        try:
-            request = GetSensorsRequest.parse_obj(payload)
-        except ValidationError as err:
-            raise InvalidRequestError(err) from err
-
-        response = await self._async_request(
-            "get",
+        response: GetSensorsResponse = await self._async_endpoint_request(
             "/sensors",
+            (
+                ("fields", fields),
+                ("location_type", location_type),
+                ("max_age", max_age),
+                ("modified_since", modified_since),
+                ("read_keys", read_keys),
+                ("show_only", sensor_indices),
+            ),
+            GetSensorsRequest,
             GetSensorsResponse,
-            params=request.dict(exclude_none=True),
         )
-        return cast(GetSensorsResponse, response)
+        return response

aiopurpleair/helpers/typing.py

@@ -3,4 +3,4 @@
 
 from pydantic import BaseModel
 
-ResponseModelT = TypeVar("ResponseModelT", bound=BaseModel)
+ModelT = TypeVar("ModelT", bound=BaseModel)

aiopurpleair/helpers/validators/__init__.py

@@ -9,6 +9,6 @@ def validate_timestamp(value: int) -> datetime:
         value: An integer (epoch datetime) to evaluate.
 
     Returns:
-        A parsed datetime.datetime object.
+        A parsed datetime.datetime object (UTC).
     """
     return datetime.utcfromtimestamp(value)