Rework sensor configuration into Pydantic models (#66)

* Rework sensor configuration into Pydantic models

* Fix linting errors

* Update README

Author: nikdoof

README.md

@@ -22,6 +22,11 @@ An external-facing, FastAPI-based front-end for accessing some Leigh Hackspace d
 | `HACKSPACE_PUBLIC_CALENDAR` | `calendar.public_events`                                                                                | The entity ID of the Home Assistant public calendar            |
 | `HACKSPACE_MEMBER_CALENDAR` | `calendar.member_events`                                                                                | The entity ID of the Home Assistant member calendar            |
 | `SENSORS_PRESSURE_ENABLED`  | `False`                                                                                                 | Enable pressure sensors                                        |
+| `SENSOR_CONFIG_FILE`        | `sensors.yaml`                                                                                          | Path to the sensors configuration                              |  |
+
+### Sensor Config File
+
+An example can be found in [`sensors.yaml`](sensors.yaml), and the details of the schema they're using can be found in `hackspaceapi.models.sensors`.
 
 ## Endpoints
 

hackspaceapi/events.py

@@ -8,7 +8,7 @@
 
 from hackspaceapi import VERSION
 
-from .config import settings
+from .models.config import settings
 from .services.homeassistant import call_homeassistant
 
 events = APIRouter()

hackspaceapi/models/__init__.py

@@ -1,8 +1,4 @@
-from pydantic import BaseModel, Field
-
-from hackspaceapi import VERSION
-
-
-class HealthResponseModel(BaseModel):
-    health: str = Field(description="State of the API", examples=["ok", "error"])
-    version: str = Field(description="Version of the API", examples=[VERSION])
+# ruff: noqa: F401
+from .config import SettingsModel
+from .health import HealthResponseModel
+from .sensors import SensorSettingsModel

hackspaceapi/config.py hackspaceapi/models/config.pyhackspaceapi/config.py renamed to hackspaceapi/models/config.py

@@ -1,21 +1,24 @@
-from pydantic import Field
+from typing import Optional
+
+from pydantic import Field, AnyHttpUrl, HttpUrl, FilePath
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from .sensors import SensorSettingsModel
 
-class Settings(BaseSettings):
+class SettingsModel(BaseSettings):
     model_config = SettingsConfigDict(env_file=".env")
 
-    base_url: str = Field(
+    base_url: AnyHttpUrl = Field(
         default="http://localhost:8000",
         description="URL base where the application will be accessible at",
     )
 
-    prometheus_instance: str = Field(
+    prometheus_instance: AnyHttpUrl = Field(
         default="http://prometheus:9090",
         description="Endpoint URL for the Prometheus instance",
     )
 
-    homeassistant_instance: str = Field(
+    homeassistant_instance: AnyHttpUrl = Field(
         default="http://homeassistant:8123",
         description="Endpoint URL for the Home Assistant instance",
     )
@@ -28,12 +31,12 @@ class Settings(BaseSettings):
         default="Leigh Hackspace", description="Name of the hackspace"
     )
 
-    hackspace_logo_url: str = Field(
+    hackspace_logo_url: HttpUrl = Field(
         default="https://raw.githubusercontent.com/leigh-hackspace/logos-graphics-assets/master/logo/rose_logo.svg",
         description="URL to the logo for the hackspace",
     )
 
-    hackspace_website_url: str = Field(
+    hackspace_website_url: HttpUrl = Field(
         default="https://leighhack.org", description="URL to the hackspace's website"
     )
 
@@ -43,7 +46,8 @@ class Settings(BaseSettings):
     )
 
     hackspace_osm_node: int = Field(
-        default=4300807520, description="OpenStreetMap Node ID for the Hackspace's location"
+        default=4300807520,
+        description="OpenStreetMap Node ID for the Hackspace's location",
     )
 
     hackspace_address_lat: float = Field(
@@ -77,5 +81,11 @@ class Settings(BaseSettings):
         default=False, description="Enable pressure sensors"
     )
 
+    sensor_config_file: Optional[FilePath] = Field(
+        default="sensors.yaml", description="Path to the sensors configuration"
+    )
+
+    sensor_config: Optional[SensorSettingsModel] = None
+
 
-settings = Settings()
+settings = SettingsModel()

hackspaceapi/models/health.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel, Field
+
+from hackspaceapi import VERSION
+
+
+class HealthResponseModel(BaseModel):
+    health: str = Field(description="State of the API", examples=["ok", "error"])
+    version: str = Field(description="Version of the API", examples=[VERSION])

hackspaceapi/models/sensors.py

@@ -0,0 +1,34 @@
+from typing import List, Optional
+
+import yaml
+from pydantic import BaseModel, Field
+
+
+class HomeassistantSensorModel(BaseModel):
+    entity: str
+    name: Optional[str] = None
+    location: str
+    unit: Optional[str] = None
+
+
+class PrometheusSensorModel(BaseModel):
+    query: str
+    name: str
+    location: Optional[str] = None
+    sensor_type: str
+    unit: Optional[str] = None
+
+
+class SensorSettingsModel(BaseModel):
+    homeassistant: List[HomeassistantSensorModel] = Field(
+        description="A list of sensors from Home Assistant"
+    )
+    prometheus: List[PrometheusSensorModel] = Field(
+        description="A list of sensors from Prometheus"
+    )
+
+    @staticmethod
+    def load_from_yaml(filename: str) -> BaseModel:
+        with open(filename, "r") as fobj:
+            data = yaml.safe_load(fobj)
+        return SensorSettingsModel(**data)

hackspaceapi/services/homeassistant.py

@@ -6,7 +6,7 @@
 from cachetools.func import ttl_cache
 from prometheus_client import Summary
 
-from hackspaceapi.config import settings
+from hackspaceapi.models.config import settings
 from hackspaceapi.services.session import get_requests_session
 
 session = get_requests_session()
@@ -28,7 +28,7 @@ def call_homeassistant(endpoint: str, **params) -> Optional[Iterable]:
     """
     Call a Home Assistant API endpoint and return the JSON if successful
     """
-    url = urljoin(settings.homeassistant_instance, endpoint)
+    url = urljoin(str(settings.homeassistant_instance), endpoint)
     try:
         resp = session.get(url, params=params)
         if resp.ok:

hackspaceapi/services/prometheus.py

@@ -6,7 +6,7 @@
 from cachetools.func import ttl_cache
 from prometheus_client import Summary
 
-from hackspaceapi.config import settings
+from hackspaceapi.models.config import settings
 from hackspaceapi.services.session import get_requests_session
 
 session = get_requests_session()
@@ -23,7 +23,7 @@ def get_prometheus_metric(query: str) -> Optional[Dict]:
     Call the configured Prometheus endpoint with a query, and return the
     resulting data if successful.
     """
-    url = urljoin(settings.prometheus_instance, "/api/v1/query")
+    url = urljoin(str(settings.prometheus_instance), "/api/v1/query")
     try:
         resp = session.get(url, params={"query": query})
         if resp.ok:

hackspaceapi/spaceapi.py

@@ -5,36 +5,14 @@
 from cachetools.func import ttl_cache
 from fastapi import APIRouter
 
-from .config import settings
+from .models.config import settings
+from .models.sensors import SensorSettingsModel
 from .services.homeassistant import get_entity_state
 from .services.prometheus import get_prometheus_metric
 from .services.website import get_membership_data
 
 spaceapi = APIRouter()
 
-# Home Assistant Sensors to export to the Space API
-# entity_id, override_name
-HOMEASSISTANT_SENSORS = (
-    ("sensor.gw_dhcp_leases_online", "WiFi Clients"),
-    ("sensor.bluetooth_proxy_temperature", "Rack 1"),
-    ("sensor.bluetooth_proxy_humidity", "Rack 1"),
-    ("sensor.pi_room_sensor_temperature", "Pi Room"),
-    ("sensor.pi_room_sensor_humidity", "Pi Room"),
-    ("sensor.pi_room_sensor_pressure", "Pi Room"),
-    ("sensor.fabrication_sensor_temperature", "Fabrication"),
-    ("sensor.fabrication_sensor_humidity", "Fabrication"),
-    ("weather.forecast_leigh_hackspace", "Outside"),
-    ("sensor.3d_1_current_state", "3D-1"),
-    ("sensor.3d_2_current_state", "3D-2"),
-    ("sensor.3d_3_current_state", "3D-3"),
-)
-
-# Prometheus queries to export to the Space API
-# query, name, sensor type
-PROMETHEUS_SENSORS = (
-    ("gocardless_members_count{}", "Active Members", "total_member_count"),
-)
-
 
 def get_state() -> dict:
     """
@@ -60,12 +38,17 @@ def get_sensors() -> dict:
     SpaceAPI response.
     """
     results = {}
-    for sensor, override_name in HOMEASSISTANT_SENSORS:
-        data = get_entity_state(sensor)
+
+    # Load the sensor config if its not initialized
+    if not settings.sensor_config:
+        settings.sensor_config = SensorSettingsModel.load_from_yaml(settings.sensor_config_file)
+
+    for sensor in settings.sensor_config.homeassistant:
+        data = get_entity_state(sensor.entity)
         if not data:
             logging.warning(
                 "Call for {0} sensor returned an empty result, skipping".format(
-                    override_name
+                    sensor.entity
                 )
             )
             continue
@@ -89,8 +72,8 @@ def get_sensors() -> dict:
             results["temperature"].append(
                 {
                     "value": value,
-                    "unit": unit_val,
-                    "location": override_name or data["attributes"]["friendly_name"],
+                    "unit": sensor.unit or unit_val,
+                    "location": sensor.location or data["attributes"]["friendly_name"],
                     "lastchange": int(arrow.get(data["last_changed"]).timestamp()),
                 }
             )
@@ -114,8 +97,8 @@ def get_sensors() -> dict:
             results["humidity"].append(
                 {
                     "value": value,
-                    "unit": unit_val,
-                    "location": override_name or data["attributes"]["friendly_name"],
+                    "unit": sensor.unit or unit_val,
+                    "location": sensor.location or data["attributes"]["friendly_name"],
                     "lastchange": int(arrow.get(data["last_changed"]).timestamp()),
                 }
             )
@@ -140,9 +123,8 @@ def get_sensors() -> dict:
                 results["barometer"].append(
                     {
                         "value": value,
-                        "unit": unit_val,
-                        "location": override_name
-                        or data["attributes"]["friendly_name"],
+                        "unit": sensor.unit or unit_val,
+                        "location": sensor.location or data["attributes"]["friendly_name"],
                         "lastchange": int(arrow.get(data["last_changed"]).timestamp()),
                     }
                 )
@@ -163,7 +145,7 @@ def get_sensors() -> dict:
             results["network_connections"].append(
                 {
                     "value": state,
-                    "location": override_name or data["attributes"]["friendly_name"],
+                    "location": sensor.location or data["attributes"]["friendly_name"],
                     "lastchange": int(arrow.get(data["last_changed"]).timestamp()),
                 }
             )
@@ -183,29 +165,29 @@ def get_sensors() -> dict:
 
             results["ext_3d_printers"].append(
                 {
-                    "name": override_name
+                    "name": sensor.name or sensor.location
                     or data["attributes"]["friendly_name"].split()[0],
                     "state": state,
                     "lastchange": int(arrow.get(data["last_changed"]).timestamp()),
                 }
             )
 
-    for query, name, sensor_type in PROMETHEUS_SENSORS:
-        data = get_prometheus_metric(query)
+    for sensor in settings.sensor_config.prometheus:
+        data = get_prometheus_metric(sensor.query)
         if not data or "result" not in data or len(data["result"]) == 0:
             logging.warning(
-                "Call for {0} sensor returned an empty result, skipping".format(name)
+                "Call for {0} sensor returned an empty result, skipping".format(sensor.name)
             )
             continue
 
-        if sensor_type not in results:
-            results[sensor_type] = []
+        if sensor.sensor_type not in results:
+            results[sensor.sensor_type] = []
 
-        if sensor_type == "total_member_count":
+        if sensor.sensor_type == "total_member_count":
             results["total_member_count"].append(
                 {
                     "value": int(data["result"][0]["value"][1]),
-                    "name": name,
+                    "name": sensor.name,
                     "lastchange": int(data["result"][0]["value"][0]),
                 }
             )
@@ -287,7 +269,7 @@ async def space_json():
             },
             "calendar": {
                 "type": "ical",
-                "url": urljoin(settings.base_url, "/events.ics"),
+                "url": urljoin(str(settings.base_url), "/events.ics"),
             },
         },
         "links": get_links(),

sensors.yaml

@@ -0,0 +1,35 @@
+homeassistant:
+  - entity: sensor.gw_dhcp_leases_online
+    location: WiFi Clients
+
+  - entity: sensor.bluetooth_proxy_temperature
+    location: Rack 1
+  - entity: sensor.bluetooth_proxy_humidity
+    location: Rack 1
+
+  - entity: sensor.pi_room_sensor_temperature
+    location: Pi Room
+  - entity: sensor.pi_room_sensor_humidity
+    location: Pi Room
+  - entity: sensor.pi_room_sensor_pressure
+    location: Pi Room
+
+  - entity: sensor.fabrication_sensor_temperature
+    location: Fabrication
+  - entity: sensor.fabrication_sensor_humidity
+    location: Fabrication
+
+  - entity: weather.forecast_leigh_hackspace
+    location: Outside
+
+  - entity: sensor.3d_1_current_state
+    location: 3D-1
+  - entity: sensor.3d_2_current_state
+    location: 3D-2
+  - entity: sensor.3d_3_current_state
+    location: 3D-3
+
+prometheus:
+  - query: gocardless_members_count{}
+    name: Active Members
+    sensor_type: total_member_count