Merge pull request #2 from rostrovsky/0.2.0

0.2.0

MANIFEST.in

@@ -0,0 +1 @@
+recursive-exclude test *

README.md

@@ -1,5 +1,9 @@
 # blexy
-Simple [OpenMetrics](https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md) exporter for BLE devices
+Simple [OpenMetrics](https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md) exporter for BLE devices.
+
+## Requirements
+* device with BLE compliant transceiver (e.g. RPi 3+)
+* Python 3.7+
 
 ## Running
 ```
@@ -31,5 +35,27 @@ ble:
       address: xx:xx:xx:xx:xx:xx
 ```
 
+## Example output
+```
+$ curl localhost:8080/metrics
+# TYPE battery_level_percentage gauge
+# UNIT battery_level_percentage percentage
+battery_level_percentage_gauge{name="living room sensor",model="LYWSD03MMC",manufacturer="Xiaomi",address="xx:xx:xx:xx:xx:xx",interface="0"} 77
+battery_level_percentage_gauge{name="bedroom sensor",model="LYWSD03MMC",manufacturer="Xiaomi",address="xx:xx:xx:xx:xx:xx",interface="0"} 88
+# TYPE humidity_percentage gauge
+# UNIT humidity_percentage percentage
+humidity_percentage_gauge{name="living room sensor",model="LYWSD03MMC",manufacturer="Xiaomi",address="xx:xx:xx:xx:xx:xx",interface="0"} 48
+humidity_percentage_gauge{name="bedroom sensor",model="LYWSD03MMC",manufacturer="Xiaomi",address="xx:xx:xx:xx:xx:xx",interface="0"} 51
+# TYPE temperature_celsius gauge
+# UNIT temperature_celsius celsius
+temperature_celsius_gauge{name="living room sensor",model="LYWSD03MMC",manufacturer="Xiaomi",address="xx:xx:xx:xx:xx:xx",interface="0"} 22.33
+temperature_celsius_gauge{name="bedroom sensor",model="LYWSD03MMC",manufacturer="Xiaomi",address="xx:xx:xx:xx:xx:xx",interface="0"} 25.11
+# TYPE voltage_volts gauge
+# UNIT voltage_volts volts
+voltage_volts_gauge{name="living room sensor",model="LYWSD03MMC",manufacturer="Xiaomi",address="xx:xx:xx:xx:xx:xx",interface="0"} 3.11
+voltage_volts_gauge{name="bedroom sensor",model="LYWSD03MMC",manufacturer="Xiaomi",address="xx:xx:xx:xx:xx:xx",interface="0"} 3.03
+# EOF
+```
+
 ## Supported devices
 * `LYWSD03MMC` (Xiaomi)

blexy/app.py

@@ -1,7 +1,9 @@
+import asyncio
 from starlette.applications import Starlette
 from starlette.responses import JSONResponse, PlainTextResponse
 from starlette.routing import Route
 from blexy.utils.config import GlobalConfig
+from blexy.utils.openmetrics import OpenMetricsAggregator
 
 
 async def homepage(request):
@@ -16,15 +18,17 @@ async def metrics(request):
     """
     Returns device readouts in OpenMetrics format
     """
-    out = []
-    for d in GlobalConfig.connected_devices():
-        if d.peripheral.waitForNotifications(5):
-            out.extend(d.open_metrics)
-    return PlainTextResponse("\n".join(out))
+    oma = OpenMetricsAggregator()
+    tasks = [d.asyncWaitForNotifications(5) for d in GlobalConfig.connected_devices()]
+    results = await asyncio.gather(*tasks)
+    for success, device in results:
+        if success:
+            oma.add_metrics(device.open_metrics)
+
+    return PlainTextResponse(oma.text)
 
 
 app = Starlette(
-    debug=True,
     routes=[
         Route("/", homepage),
         Route("/metrics", metrics),

blexy/devices/LYWSD03MMC.py

@@ -1,6 +1,6 @@
-from bluepy import btle
 from typing import List
 from blexy.devices.abstract_device import AbstractDevice
+from blexy.utils.openmetrics import OpenMetric, OpenMetricType
 
 
 class LYWSD03MMC(AbstractDevice):
@@ -19,20 +19,20 @@ def connect(self):
         if self.is_connected:
             return self
 
-        print(f"Connecting {self.name} ({self.model})")
+        self.log.info("Connecting...")
         self.peripheral.connect(self.address)
         self.peripheral.writeCharacteristic(
             0x0038, b"\x01\x00", True
         )  # enable notifications of Temperature, Humidity and Battery voltage
         self.peripheral.writeCharacteristic(0x0046, b"\xf4\x01\x00", True)
         self.peripheral.withDelegate(self)
-        print(f"Connected {self.name} ({self.model})")
+        self.log.info(f"Connected")
         self.is_connected = True
         return self
 
     def handleNotification(self, cHandle, data):
         try:
-            print(f"{self.name} ({self.model}) : received {data}")
+            self.log.info(f"received data: {data.hex()}")
             self.temperature = (
                 int.from_bytes(data[0:2], byteorder="little", signed=True) / 100
             )
@@ -42,14 +42,38 @@ def handleNotification(self, cHandle, data):
                 int(round((self.voltage - 2.1), 2) * 100), 100
             )  # 3.1 or above --> 100% 2.1 --> 0 %
         except Exception as e:
-            print("e")
+            self.log.error(e)
 
     @property
-    def open_metrics(self) -> List[str]:
+    def open_metrics(self) -> List[OpenMetric]:
         output = [
-            f"temperature{self._open_metrics_labels} {self.temperature}",
-            f"humidity{self._open_metrics_labels} {self.humidity}",
-            f"voltage{self._open_metrics_labels} {self.voltage}",
-            f"battery_level{self._open_metrics_labels} {self.battery_level}",
+            OpenMetric(
+                name="temperature",
+                type=OpenMetricType.gauge,
+                unit="celsius",
+                labels=self._open_metrics_labels,
+                value=self.temperature,
+            ),
+            OpenMetric(
+                name="humidity",
+                type=OpenMetricType.gauge,
+                unit="percentage",
+                labels=self._open_metrics_labels,
+                value=self.humidity,
+            ),
+            OpenMetric(
+                name="voltage",
+                type=OpenMetricType.gauge,
+                unit="volts",
+                labels=self._open_metrics_labels,
+                value=self.voltage,
+            ),
+            OpenMetric(
+                name="battery_level",
+                type=OpenMetricType.gauge,
+                unit="percentage",
+                labels=self._open_metrics_labels,
+                value=self.battery_level,
+            ),
         ]
         return output

blexy/devices/abstract_device.py

@@ -1,7 +1,10 @@
 import json
-from abc import ABCMeta, abstractmethod, abstractproperty
-from typing import List
+from abc import ABCMeta, abstractmethod
+from typing import List, Tuple
 from bluepy import btle
+from twiggy import log
+
+from blexy.utils.concurrency import run_in_executor
 
 
 class AbstractDevice(btle.DefaultDelegate, metaclass=ABCMeta):
@@ -13,6 +16,10 @@ def __init__(self, name, address, interface):
         self.peripheral = btle.Peripheral(deviceAddr=None, iface=self.interface)
         self.is_connected = False
 
+    @property
+    def log(self):
+        return log.name(f"{self.name}@{self.address}")
+
     @abstractmethod
     def connect(self) -> "AbstractDevice":
         pass
@@ -21,13 +28,28 @@ def connect(self) -> "AbstractDevice":
     def handleNotification(self, cHandle, data):
         pass
 
-    @abstractproperty
+    @run_in_executor
+    def __asyncWaitForNotifications(self, timeout) -> Tuple[bool, "AbstractDevice"]:
+        return self.peripheral.waitForNotifications(timeout), self
+
+    async def asyncWaitForNotifications(self, timeout) -> Tuple[bool, "AbstractDevice"]:
+        self.log.debug(f"waiting for notification with timeout {timeout}s")
+        return await self.__asyncWaitForNotifications(timeout), self
+
+    @property
+    @abstractmethod
     def open_metrics(self) -> List[str]:
         pass
 
     @property
-    def _open_metrics_labels(self) -> str:
-        return f'{{name="{self.name}",model="{self.model}",manufacturer="{self.manufacturer}",address="{self.address}",interface="{self.interface}"}}'
+    def _open_metrics_labels(self) -> dict:
+        return {
+            "name": self.name,
+            "model": self.model,
+            "manufacturer": self.manufacturer,
+            "address": self.address,
+            "interface": self.interface,
+        }
 
     @property
     def as_dict(self) -> dict:

blexy/main.py

@@ -2,6 +2,7 @@
 import uvicorn
 import blexy.app
 from blexy.utils.config import GlobalConfig
+from twiggy import quick_setup, levels
 
 
 @click.command()
@@ -16,6 +17,7 @@ def cli(port, config_file, log_level):
     GlobalConfig.load_from_file(config_file)
     app_port = port if port else GlobalConfig.port
     app_log_level = log_level if log_level else GlobalConfig.log_level
+    quick_setup(min_level=getattr(levels, app_log_level.upper()))
     GlobalConfig.connect_all_devices()
     uvicorn.run(blexy.app.app, host="0.0.0.0", port=app_port, log_level=app_log_level)
 

blexy/utils/concurrency.py

@@ -0,0 +1,11 @@
+import functools
+import asyncio
+
+
+def run_in_executor(f):
+    @functools.wraps(f)
+    def inner(*args, **kwargs):
+        loop = asyncio.get_running_loop()
+        return loop.run_in_executor(None, lambda: f(*args, **kwargs))
+
+    return inner

blexy/utils/config.py

@@ -1,8 +1,10 @@
-from abc import abstractproperty
+import yaml
+import concurrent.futures
+
 from pathlib import Path
 from importlib import import_module
 from typing import List
-import yaml
+from twiggy import log
 
 from blexy.devices.abstract_device import AbstractDevice
 
@@ -12,9 +14,11 @@ class GlobalConfig:
     log_level = None
     devices = None
     _device_objects = None
+    log = log.name("GlobalConfig")
 
     @staticmethod
     def load_from_file(file_path: str) -> "GlobalConfig":
+        GlobalConfig.log.info(f"Loading config from: {file_path}")
         fp = Path(file_path)
         with open(fp, "r") as cf:
             config = yaml.load(cf, yaml.SafeLoader)
@@ -38,14 +42,18 @@ def load_from_file(file_path: str) -> "GlobalConfig":
                 )
                 GlobalConfig._device_objects.append(device_obj)
             except Exception as e:
-                print(f'could not import device "{dev_name}" ({dev_model})')
-                print(e)
+                GlobalConfig.log.error(
+                    f'could not import device "{dev_name}" ({dev_model})'
+                )
+                GlobalConfig.log.error(e)
 
     @staticmethod
     def connected_devices() -> List[AbstractDevice]:
         return [d for d in GlobalConfig._device_objects if d.is_connected]
 
     @staticmethod
     def connect_all_devices():
-        for d in GlobalConfig._device_objects:
-            d.connect()
+        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
+            tasks = [executor.submit(d.connect) for d in GlobalConfig._device_objects]
+            for _ in concurrent.futures.as_completed(tasks):
+                pass

blexy/utils/openmetrics.py

@@ -0,0 +1,76 @@
+from typing import Any, List
+from dataclasses import dataclass
+from itertools import groupby
+
+
+class OpenMetricType:
+    gauge = "gauge"
+    counter = "counter"
+    state_set = "state_set"
+    info = "info"
+    histogram = "histogram"
+    gauge_histogram = "gauge_histogram"
+    summary = "summary"
+    unknown = "unknown"
+
+
+@dataclass
+class OpenMetric:
+    name: str
+    type: str
+    unit: str
+    labels: dict
+    value: Any
+
+    @property
+    def labels_text(self):
+        out = []
+        for k, v in self.labels.items():
+            out.append(f'{k}="{v}"')
+        return ",".join(out)
+
+    @property
+    def text(self):
+        return f"{self.name}_{self.unit}_{self.type}{{{self.labels_text}}} {self.value}"
+
+    @property
+    def type_metadata(self):
+        return f"# TYPE {self.name}_{self.unit} {self.type}"
+
+    @property
+    def unit_metadata(self):
+        return f"# UNIT {self.name}_{self.unit} {self.unit}"
+
+
+class OpenMetricsAggregator:
+    def __init__(self):
+        self.metrics = []
+
+    def add_metric(self, metric: OpenMetric) -> "OpenMetricsAggregator":
+        self.metrics.append(metric)
+        return self
+
+    def add_metrics(self, metrics: List[OpenMetric]) -> "OpenMetricsAggregator":
+        self.metrics.extend(metrics)
+        return self
+
+    @property
+    def grouped_metrics(self):
+        sorted_metrics = sorted(
+            self.metrics, key=lambda m: f"{m.name}_{m.unit}_{m.type}"
+        )
+        return groupby(sorted_metrics, lambda m: m.name)
+
+    @property
+    def text(self):
+        out = []
+        for metric_name, group in self.grouped_metrics:
+            # out.append(f"{group}: {metric_name}")
+            for metric in group:
+                if metric.type_metadata not in out:
+                    out.append(metric.type_metadata)
+                if metric.unit_metadata not in out:
+                    out.append(metric.unit_metadata)
+                out.append(str(metric.text))
+        out.append("# EOF")
+        return "\n".join(out)

config.yaml

@@ -4,4 +4,7 @@ ble:
   devices:
     - name: living room sensor
       model: LYWSD03MMC
-      address: A4:C1:38:64:49:02
+      address: A4:C1:38:64:49:02
+    - name: bedroom sensor
+      model: LYWSD03MMC
+      address: A4:C1:38:9B:16:5B

requirements.txt

@@ -2,4 +2,5 @@ starlette
 pyyaml
 bluepy
 uvicorn
-click
+click
+twiggy