Implement configuration via Pydantic & change project structure

This commit is contained in:
Denis-Cosmin Nutiu 2024-02-06 23:22:47 +02:00
parent 33b3432787
commit c4b28a56e6
10 changed files with 215 additions and 47 deletions

View file

@ -1,4 +1,3 @@
# Introduction
Simple script utilities to add BME680 sensor readings to Apple Homekit using a Raspberry PI with minimal configuration.
@ -21,18 +20,29 @@ sudo apt-get install libavahi-compat-libdnssd-dev
pip3 install -r requirements.txt
```
### Sensors
### Application
The `sensors` directory contains code for operating the bme680 sensor.
The `app` directory contains the source code of this application
The sensor values are collected and exposed in HomeKit and as prometheus metrics.
The prometheus metrics can be accessed on port `8000`.
Sensor values are collected and exposed in HomeKit and as prometheus metrics.
By default, metrics can be accessed on port `8000`.
### Configuration
Before running the program edit the `config.yaml` file and replace at least `persist_file` field.
The program will search for the `config.yaml` file in the current directory but this can be configured to another directory
by settings the `HOMEKIT_CONFIG` environment variable.
### Running
Run the program once to pair it with your ios. ex:
```bash
cd sensors
python3 main.py
python3 -m app.main
Setup payload: X-HM://0023K50QET2YB
Scan this code with your HomeKit app on your iOS device:
@ -40,10 +50,19 @@ Or enter this code in your HomeKit app on your iOS device: 053-86-998
```
### SystemD Service
Edit `./systemd/bme680-homekit.service` and replace relevant variables such as `WorkingDirectory`, `ExecStart`
and `User`.
Run `sudo ./systemd/install.sh` to install SystemD service automatically on your system.
Or follow the manual steps.
Copy the systemd service.
```bash
sudo cp bme680-homekit.service /etc/systemd/system
sudo cp ./systemd/bme680-homekit.service /etc/systemd/system
sudo systemctl status bme680-homekit
```
@ -54,6 +73,7 @@ sudo systemctl status bme680-homekit
```
Start the service
```bash
sudo systemctl start bme680-homekit
sudo systemctl status bme680-homekit

0
app/__init__.py Normal file
View file

117
app/config.py Normal file
View file

@ -0,0 +1,117 @@
import functools
import os
from pathlib import Path
from typing import Type, Tuple, Any, Dict
import yaml
from pydantic import BaseModel, Field
from pydantic.fields import FieldInfo
from pydantic_settings import BaseSettings, PydanticBaseSettingsSource
config_file_default_location = "config.yaml"
class PrometheusSettings(BaseModel):
enabled: bool = Field(True, description="Enable prometheus metrics server.")
port: int = Field(8000, description="The prometheus metrics server port.")
class Bme680Settings(BaseModel):
enabled: bool = Field(True, description="If the sensor should be enabled on not.")
address: int = Field(
0x76, description="The I2C address of the sensor. Defaults to primary"
)
name: str = Field("Climate Sensor", description="The name of the sensor.")
class BridgeSettings(BaseModel):
display_name: str = Field(
"Bridge", description="The display name of the HAP bridge."
)
bme680: Bme680Settings = Field(
Bme680Settings(), description="Settings for the BME680 module."
)
class HomekitAccessoryProtocolSettings(BaseModel):
port: int = Field(
51826, description="The port of the homekit accessory protocol server."
)
persist_file: str = Field(
...,
description="The persistence file which holds the homekit accessory protocol server's state.",
)
bridge: BridgeSettings = Field(
BridgeSettings(), description="The HAP's default bridge settings."
)
class YamlConfigSettingsSource(PydanticBaseSettingsSource):
"""
A simple settings source class that loads variables from a YAML file
at the project's root.
Here we happen to choose to use the `env_file_encoding` from Config
when reading `config.yaml`
"""
@functools.lru_cache
def read_file_content(self):
encoding = self.config.get("env_file_encoding")
return yaml.safe_load(
Path(
os.getenv("HOMEKIT_CONFIG", default=config_file_default_location)
).read_text(encoding)
)
def get_field_value(
self, field: FieldInfo, field_name: str
) -> Tuple[Any, str, bool]:
file_content_json = self.read_file_content()
field_value = file_content_json.get(field_name)
return field_value, field_name, False
def prepare_field_value(
self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool
) -> Any:
return value
def __call__(self) -> Dict[str, Any]:
d: Dict[str, Any] = {}
for field_name, field in self.settings_cls.model_fields.items():
field_value, field_key, value_is_complex = self.get_field_value(
field, field_name
)
field_value = self.prepare_field_value(
field_name, field, field_value, value_is_complex
)
if field_value is not None:
d[field_key] = field_value
return d
class Settings(BaseSettings):
prometheus: PrometheusSettings = Field(
PrometheusSettings(), description="Settings for prometheus."
)
hap: HomekitAccessoryProtocolSettings = Field(
..., description="Homekit Accessory Protocol server settings."
)
@classmethod
def settings_customise_sources(
cls,
settings_cls: Type[BaseSettings],
init_settings: PydanticBaseSettingsSource,
env_settings: PydanticBaseSettingsSource,
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> Tuple[PydanticBaseSettingsSource, ...]:
return (
init_settings,
YamlConfigSettingsSource(settings_cls),
env_settings,
file_secret_settings,
)

41
app/main.py Normal file
View file

@ -0,0 +1,41 @@
import signal
from prometheus_client import start_http_server
from pyhap.accessory import Bridge
from pyhap.accessory_driver import AccessoryDriver
from app.config import Settings
from app.sensors.bme import Bme680Sensor
def get_bridge(accessory_driver: AccessoryDriver, settings: Settings):
"""
Returns a Homekit Bridge.
Parameters
----------
accessory_driver: AccessoryDriver
The accessory driver.
"""
bridge = Bridge(accessory_driver, settings.hap.bridge.display_name)
if settings.hap.bridge.bme680.enabled:
bridge.add_accessory(
Bme680Sensor(
accessory_driver, settings.hap.bridge.bme680.name, settings=settings
)
)
return bridge
if __name__ == "__main__":
settings = Settings()
# Start prometheus metrics server. Any metrics will be registered automatically.
if settings.prometheus.enabled:
start_http_server(settings.prometheus.port)
# Python HAP
driver = AccessoryDriver(
port=settings.hap.port, persist_file=settings.hap.persist_file
)
driver.add_accessory(accessory=get_bridge(driver, settings))
signal.signal(signal.SIGTERM, driver.signal_handler)
driver.start()

0
app/sensors/__init__.py Normal file
View file

View file

@ -1,21 +1,12 @@
import logging
import os
import signal
import sys
import bme680
from prometheus_client import Gauge, start_http_server
from pyhap.accessory import Accessory, Bridge
from pyhap.accessory_driver import AccessoryDriver
from prometheus_client import Gauge
from pyhap.accessory import Accessory
from pyhap.const import CATEGORY_SENSOR
def fail(message: str):
"""
Fail crashes the program and logs the message.
"""
logging.critical(message)
sys.exit(1)
from app.config import Settings
class Bme680Sensor(Accessory):
@ -23,14 +14,17 @@ class Bme680Sensor(Accessory):
category = CATEGORY_SENSOR # This is for the icon in the iOS Home app.
def __init__(self, *args, **kwargs):
def __init__(self, driver, display_name, *, aid=None, settings: Settings):
"""Here, we just store a reference to the current temperature characteristic and
add a method that will be executed every time its value changes.
"""
# If overriding this method, be sure to call the super's implementation first.
super().__init__(*args, **kwargs)
super().__init__(driver, display_name, aid=aid)
self.sensor = bme680.BME680(bme680.I2C_ADDR_PRIMARY)
self.settings = settings
self.sensor = bme680.BME680(
settings.hap.bridge.bme680.address or bme680.I2C_ADDR_PRIMARY
)
self.sensor.set_humidity_oversample(bme680.OS_2X)
self.sensor.set_pressure_oversample(bme680.OS_4X)
@ -87,28 +81,11 @@ class Bme680Sensor(Accessory):
self._run()
except IOError as e:
# This happens from time to time, best we stop and let systemd restart us.
fail(f"Failed due to IOError: {e}")
logging.critical("Failed to run BME680.")
sys.exit(1)
def stop(self):
"""We override this method to clean up any resources or perform final actions, as
this is called by the AccessoryDriver when the Accessory is being stopped.
"""
print("Stopping accessory.")
def get_bridge(accessory_driver):
bridge = Bridge(accessory_driver, "Bridge")
bridge.add_accessory(Bme680Sensor(accessory_driver, "Sensor"))
return bridge
if __name__ == "__main__":
# Prometheus' metrics server
start_http_server(8000)
# Python HAP
driver = AccessoryDriver(
port=51826, persist_file="/home/pi/bme680-homekit/sensors/accessory.state"
)
driver.add_accessory(accessory=get_bridge(driver))
signal.signal(signal.SIGTERM, driver.signal_handler)
driver.start()

12
config.yaml Normal file
View file

@ -0,0 +1,12 @@
prometheus:
enabled: yes
port: 8000
hap:
port: 51826
persist_file: /home/pi/bme680-homekit/sensors/accessory.state
bridge:
display_name: Bridge
bme680:
enabled: yes
address: 118 # Primary I2C Address
name: "Climate Sensor"

View file

@ -1,3 +1,4 @@
bme680==1.1.1
HAP-python==4.4.0
prometheus-client==0.14.1
prometheus-client== 0.19.0
PyYAML==6.0.1

View file

@ -7,7 +7,7 @@ WorkingDirectory=/home/pi/bme680-homekit/
Restart=on-failure
RestartSec=5s
User=pi
ExecStart=/usr/bin/python3 /home/pi/bme680-homekit/sensors/main.py
ExecStart=/usr/bin/python3 -m app.main
[Install]
WantedBy=multi-user.target

2
sensors/install.sh → systemd/install.sh Executable file → Normal file
View file

@ -1,5 +1,5 @@
# Install the systemd service
cp bme680-homekit.service /etc/systemd/system
cp ./systemd/bme680-homekit.service /etc/systemd/system
systemctl daemon-reload
systemctl start bme680-homekit
systemctl enable bme680-homekit