Prometheus exporters for various air sensors

prometheus-bmp280-exporter.py

#!/usr/bin/env python3

import iio
import datetime
import logging
import time
from prometheus_client import start_http_server, Gauge

GAUGE_PRESSURE = Gauge('pressure',
                       'Temperature (kPa)',
                       namespace='air',
                       unit='kilopascal')
GAUGE_TEMP = Gauge('temp',
                   'Temperature (Celsius)',
                   namespace='air',
                   unit='celsius')

ctx = iio.Context()
dev = ctx.find_device('bmp280')
pressure = dev.find_channel('pressure')
temp = dev.find_channel('temp')

start_http_server(9356, '127.0.0.1')

start = time.monotonic()

logging.basicConfig(level=logging.INFO)

while True:
    p = float(pressure.attrs['input'].value)
    t = float(temp.attrs['input'].value) / 1000
    logging.info('%f kPa, %f celcius', p, t)
    GAUGE_PRESSURE.set(p)
    GAUGE_TEMP.set(t)

    # event.timestamp_ns seems to be relative to system boot
    now = datetime.datetime.now(datetime.UTC)

    time.sleep(1.0 - ((time.monotonic() - start) % 1.0))

prometheus-cjco-uart-exporter.py

#!/usr/bin/env python3

import serial
import sys
import time
import logging
import datetime
from prometheus_client import start_http_server, Gauge

GAUGE_CO = Gauge('co', 'CO concentration (PPM)', namespace='air', unit='ppm')

logging.basicConfig(level=logging.INFO)
logging.info('Preheating!')
time.sleep(3)

# Usage: ./prometheus-cjco-uart-exporter.py /dev/ttyUSB0 9381 127.0.0.1
_, serdev, port, addr = sys.argv

start_http_server(int(port), addr)

with serial.Serial(serdev, 9600, timeout=2) as ser:
    ser.reset_input_buffer()
    while d := ser.read(6):
        logging.debug('RX: %s', d)
        id_, comsb, colsb, fsmsb, fslsb, checksum = d

        assert id_ == 0x2c, f'Bad device id! {d}'
        assert checksum == sum(d[:-1]) & 0xff, f'Bad checksum! {d}'

        co = (comsb << 8) + colsb
        fullscale = (fsmsb << 8) + fslsb

        assert fullscale == 1000, f'Bad fullscale! {d}'

        logging.info('CO: %d ppm', co)
        now = datetime.datetime.now(datetime.UTC)

        GAUGE_CO.set(co)

prometheus-cjsh2x-exporter.py

#!/usr/bin/env python3
# CJ-SH20, CJ-SH22

import serial
import sys
import time
import logging
import datetime
from prometheus_client import start_http_server, Gauge

GAUGE_CH2O = Gauge('ch2o',
                   'CH2O concentration (PPB)',
                   namespace='air',
                   unit='ppb')

logging.basicConfig(level=logging.INFO)
logging.info('Preheating!')
time.sleep(180)

# Usage: ./prometheus-cjsh2x-exporter.py /dev/ttyUSB0 9384 127.0.0.1
_, serdev, port, addr = sys.argv

start_http_server(int(port), addr)

with serial.Serial(serdev, 9600, timeout=3) as ser:
    ser.reset_input_buffer()
    while d := ser.read(9):
        logging.debug('RX: %s', d)
        sof, id_, unit, _, msb, lsb, fsmsb, fslsb, checksum = d

        assert sof == 0xff
        assert id_ == 0x17
        assert unit == 0x04
        assert checksum == (((sum(d[1:-1]) & 0xff) ^ 0xff) + 1) & 0xff

        ch2o = (msb << 8) + lsb
        fullscale = (fsmsb << 8) + fslsb

        assert fullscale == 0x1388

        logging.info('CH2O: %d ppb', ch2o)
        now = datetime.datetime.now(datetime.UTC)

        GAUGE_CH2O.set(ch2o)

prometheus-jw01co2-exporter.py

#!/usr/bin/env python3

import serial
import sys
import time
import logging
import datetime
from prometheus_client import start_http_server, Gauge

GAUGE_CO2 = Gauge('co2',
                  'CO2 concentration (PPM)',
                  namespace='air',
                  unit='ppm')

logging.basicConfig(level=logging.INFO)
logging.info('Preheating!')
time.sleep(60)

# Usage: ./prometheus-jw01co2-exporter.py /dev/ttyUSB1 9325 127.0.0.1
_, serdev, port, addr = sys.argv

start_http_server(int(port), addr)

with serial.Serial(serdev, 9600, timeout=2) as ser:
    ser.reset_input_buffer()
    while d := ser.read(6):
        logging.debug('RX: %s', d)
        id_, comsb, colsb, fsmsb, fslsb, checksum = d

        assert id_ == 0x2c, f'Bad device id! {d}'
        assert checksum == sum(d[:-1]) & 0xff, f'Bad checksum! {d}'

        co2 = (comsb << 8) + colsb
        fullscale = (fsmsb << 8) + fslsb

        assert fullscale == 1023, f'Bad fullscale! {d}'

        logging.info('CO2: %d ppm', co2)
        now = datetime.datetime.now(datetime.UTC)

        GAUGE_CO2.set(co2)

prometheus-scd30-i2c-exporter.py

#!/usr/bin/env python3

from smbus2 import SMBus, i2c_msg
import sys
import time
import struct
import logging
import itertools
from prometheus_client import start_http_server, Gauge

GAUGE_CO2 = Gauge('co2',
                  'CO2 concentration (PPM)',
                  namespace='air',
                  unit='ppm')
GAUGE_TEMP = Gauge('temp',
                   'Temperature (Celsius)',
                   namespace='air',
                   unit='celsius')
GAUGE_HUMID = Gauge('humid', 'Humidity (%)', namespace='air')

ADDR = 0x61


def crc8(msg: bytes) -> int:
    crc = 0xff  # Initialization
    for byte in msg:
        crc ^= byte
        for _ in range(8):
            if crc & 0x80:
                crc = (crc << 1) ^ 0x31  # Polynomial
            else:
                crc <<= 1
        crc &= 0xff  # Ensure the result stays within 8 bits
    return crc ^ 0x00  # Final XOR


def checkcrc(msg: bytes) -> bool:
    for chunk in itertools.batched(msg, 3):
        if crc8(chunk[:-1]) != chunk[-1]:
            return False
    return True


# SCD30 wants a STOP condition instead of a RESTART between WRITE and READ
def i2cread(bus: SMBus, cmd: bytes, bytestoread: int) -> bytes:
    write = i2c_msg.write(ADDR, cmd)
    bus.i2c_rdwr(write)

    read = i2c_msg.read(ADDR, bytestoread)
    bus.i2c_rdwr(read)
    return bytes(read)


def loop(bus: SMBus):
    start = time.monotonic()
    while True:
        time.sleep(2.0 - ((time.monotonic() - start) % 2.0))

        data = i2cread(bus, [0x02, 0x02], 3)
        logging.debug('Get data ready status: %s', data)
        assert checkcrc(data), f'Bad CRC! {data}'

        if (data[0] << 8) + data[1] != 1:
            logging.warning('Data not ready! %s', data)
            continue

        data = i2cread(bus, [0x03, 0x00], 18)
        logging.debug('Read measurement: %s', data)
        assert checkcrc(data), f'Bad CRC! {data}'

        co2bytes = data[0:2] + data[3:5]
        co2 = struct.unpack('>f', co2bytes)[0]
        GAUGE_CO2.set(co2)

        tempbytes = data[6:8] + data[9:11]
        temp = struct.unpack('>f', tempbytes)[0]
        GAUGE_TEMP.set(temp)

        humidbytes = data[12:14] + data[15:17]
        humid = struct.unpack('>f', humidbytes)[0]
        GAUGE_HUMID.set(humid)

        logging.info('CO2: %.2f ppm, %.2f °C, RH: %.2f%%', co2, temp, humid)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)

    # Usage: ./prometheus-scd30-i2c-exporter.py /dev/i2c-1 9316 127.0.0.1
    _, busdev, port, addr = sys.argv

    start_http_server(int(port), addr)
    with SMBus(busdev) as bus:
        loop(bus)

prometheus-sps30-uart-exporter.py

#!/usr/bin/env python3

import sys
import serial
import time
import struct
import logging
from prometheus_client import start_http_server, Gauge

GAUGE_PM1 = Gauge('pm1', 'PM 1.0 (ug/m3)', namespace='air')
GAUGE_PM25 = Gauge('pm25', 'PM 2.5 (ug/m3)', namespace='air')
GAUGE_PM4 = Gauge('pm4', 'PM 4.0 (ug/m3)', namespace='air')
GAUGE_PM10 = Gauge('pm10', 'PM 10 (ug/m3)', namespace='air')


def checksum(frame: bytes) -> int:
    lsb = sum(frame) & 0xff
    return 0xff ^ lsb


ESCAPE_TABLE = {
    0x7e: [0x7d, 0x5e],
    0x7d: [0x7d, 0x5d],
    0x11: [0x7d, 0x31],
    0x13: [0x7d, 0x33],
}


def escape(msg: bytes) -> bytes:
    result = []
    for b in msg:
        if b in ESCAPE_TABLE:
            result += ESCAPE_TABLE[b]
        else:
            result.append(b)
    return result


def unescape(msg: bytes) -> bytes:
    result = []
    i = 0
    while i < len(msg):
        a = msg[i]
        if a != 0x7d:
            result.append(a)
            i += 1
            continue
        b = msg[i + 1]
        c = next(k for k, v in ESCAPE_TABLE.items() if v == [a, b])
        result.append(c)
        i += 2
    return bytes(result)


def mosi(uart: serial.Serial, cmd: int, txdata: bytes):
    frame = [0x00, cmd, len(txdata)] + txdata
    chk = checksum(frame)
    escaped = escape(frame + [chk])
    uart.write([0x7e] + escaped + [0x7e])


def miso(uart: serial.Serial):
    msg = uart.read(100)
    logging.debug('MISO: %s', msg)
    unescaped = unescape(msg)
    sof = unescaped[0]
    adr = unescaped[1]
    cmd = unescaped[2]
    state = unescaped[3]
    length = unescaped[4]
    rxdata = unescaped[5:-2]
    chk = unescaped[-2]
    eof = unescaped[-1]
    assert (sof, adr, eof) == (0x7e, 0x00, 0x7e)
    assert state == 0x00
    assert length == len(rxdata)
    frame = bytes([adr, cmd, state, length]) + rxdata
    assert checksum(frame) == chk
    return cmd, rxdata


START_MEASUREMENT = 0x00
STOP_MEASUREMENT = 0x01
READ_MEASURED_VALUES = 0x03
READ_WRITE_AUTO_CLEANING_INTERVAL = 0x80
START_FAN_CLEANING = 0x56
DEVICE_RESET = 0xd3


def read_meas(uart: serial.Serial) -> tuple[float, float, float, float]:
    mosi(uart, READ_MEASURED_VALUES, [])
    _, rxdata = miso(uart)
    if not rxdata:
        logging.info('No new measurements are available')
        return (None, None, None, None)
    logging.debug('READ_MEASURED_VALUES: %s', rxdata)
    pm1 = struct.unpack('>f', rxdata[0:4])[0]
    pm25 = struct.unpack('>f', rxdata[4:8])[0]
    pm4 = struct.unpack('>f', rxdata[8:12])[0]
    pm10 = struct.unpack('>f', rxdata[12:16])[0]
    logging.info('PM1.0: %f, PM2.5: %f, PM4.0: %f, PM10: %f', pm1, pm25, pm4, pm10)
    return (pm1, pm25, pm4, pm10)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)

    # Usage: ./prometheus-sps30-uart-exporter.py /dev/ttyAMA2 9378 127.0.0.1
    _, serdev, port, addr = sys.argv
    start_http_server(int(port), addr)

    # Max. Response Time = 20ms
    with serial.Serial(serdev, 115200, timeout=0.05) as ser:
        ser.reset_input_buffer()

        mosi(ser, DEVICE_RESET, [])
        logging.info('DEVICE_RESET: %s', miso(ser))

        mosi(ser, START_MEASUREMENT, [0x01, 0x03])
        logging.info('START_MEASUREMENT: %s', miso(ser))

        logging.info('Preheating!')
        time.sleep(30)

        start = time.monotonic()
        try:
            while True:
                time.sleep(1.0 - ((time.monotonic() - start) % 1.0))
                pm1, pm25, pm4, pm10 = read_meas(ser)
                GAUGE_PM1.set(pm1)
                GAUGE_PM25.set(pm25)
                GAUGE_PM4.set(pm4)
                GAUGE_PM10.set(pm10)
        finally:
            mosi(ser, STOP_MEASUREMENT, [0x01, 0x03])
            logging.info('STOP_MEASUREMENT: %s', miso(ser))

prometheus-xl0801-exporter.py

#!/usr/bin/env python3
"""
Usage:

    ./prometheus-xl0801-exporter.py 9341 127.0.0.1

This exporter uses explicit timestamps. You should enable honor_timestamps in your Prometheus config:

    scrape_configs:
    - job_name: xl0801
      scrape_interval: 1s
      honor_timestamps: true
      static_configs:
      - targets:
        - localhost:9341
"""

import sys
import asyncio
from datetime import datetime
from dataclasses import dataclass
from bleak import BleakScanner
from bleak.args.bluez import BlueZScannerArgs, OrPattern, AdvertisementDataType
from prometheus_client import start_http_server
from prometheus_client.core import GaugeMetricFamily, REGISTRY
from prometheus_client.registry import Collector

COLORS = {
    'ED:68:01:11:11:11': 0x73bf69,
    'ED:68:01:22:22:22': 0xf2cc0c,
    'ED:68:01:33:33:33': 0x5794f2,
    'ED:68:01:44:44:44': 0xff9830,
    'ED:68:01:55:55:55': 0xf2495c,
    'ED:68:01:66:66:66': 0xb877d9,
}


@dataclass
class Sample:
    temp: float
    humid: int
    time: datetime


samples: dict[str, Sample] = {}

# The sensor broadcasts at irregular interval
# https://prometheus.io/docs/specs/om/open_metrics_spec/
#     MetricPoints SHOULD NOT have explicit timestamps
# but we have good reason to use it


# A custom Collector is needed to set explicit timestamp
class CustomCollector(Collector):

    def collect(self):
        t = GaugeMetricFamily('air_temp',
                              'Temperature (Celsius)',
                              labels=['device'],
                              unit='celsius')
        for mac, sample in samples.items():
            t.add_metric([mac], sample.temp, timestamp=sample.time.timestamp())
        yield t

        h = GaugeMetricFamily('air_humid', 'Humidity (%)', labels=['device'])
        for mac, sample in samples.items():
            h.add_metric([mac],
                         sample.humid,
                         timestamp=sample.time.timestamp())
        yield h


REGISTRY.register(CustomCollector())

# scanning_mode='passive'
ARGS = BlueZScannerArgs(or_patterns=[
    OrPattern(0, AdvertisementDataType.MANUFACTURER_SPECIFIC_DATA, b'\x01\x09')
])


async def main():
    async with BleakScanner(scanning_mode='passive', bluez=ARGS) as scanner:
        async for device, adv in scanner.advertisement_data():
            if device.address.startswith(
                    'ED:68:01') and device.name == 'XL0801':
                now = datetime.now()
                r, g, b = COLORS.get(device.address, 0x9b9b9b).to_bytes(3)

                for data in adv.manufacturer_data.values():
                    temp = (data[0] * 256 + data[1]) / 10
                    humid = data[2]
                    print(
                        f'{now}, {temp:.1f} °C, RH: {humid}%, device: \x1b[38;2;{r};{g};{b}m{device.address}\x1b[0m'
                    )
                    samples[device.address] = Sample(temp=temp,
                                                     humid=humid,
                                                     time=now)


if __name__ == '__main__':
    _, port, addr = sys.argv

    start_http_server(int(port), addr)
    asyncio.run(main())

prometheus_micropython_sps30_exporter.py

# Usage:
#
#     from prometheus_micropython_sps30_exporter import start_http_server
#
#     # Connect WiFi, e.g. https://docs.micropython.org/en/latest/esp32/quickref.html#wlan
#     do_connect()
#     asyncio.run(start_http_server())

import machine
import socket
import struct
import asyncio


def checksum(frame: bytes) -> int:
    lsb = sum(frame) & 0xff
    return 0xff ^ lsb


ESCAPE_TABLE = {
    0x7e: [0x7d, 0x5e],
    0x7d: [0x7d, 0x5d],
    0x11: [0x7d, 0x31],
    0x13: [0x7d, 0x33],
}


def escape(msg: bytes) -> bytes:
    result = []
    for b in msg:
        if b in ESCAPE_TABLE:
            result += ESCAPE_TABLE[b]
        else:
            result.append(b)
    return result


def unescape(msg: bytes) -> bytes:
    result = []
    i = 0
    while i < len(msg):
        a = msg[i]
        if a != 0x7d:
            result.append(a)
            i += 1
            continue
        b = msg[i + 1]
        c = next(k for k, v in ESCAPE_TABLE.items() if v == [a, b])
        result.append(c)
        i += 2
    return bytes(result)


async def mosi(uart: asyncio.Stream, cmd: int, txdata: bytes):
    frame = [0x00, cmd, len(txdata)] + txdata
    chk = checksum(frame)
    escaped = escape(frame + [chk])
    b = bytes([0x7e] + escaped + [0x7e])
    # print('[DEBUG] writing', b)
    uart.write(b)
    await uart.drain()


async def miso(uart: asyncio.Stream):
    # await asyncio.sleep_ms(50)
    msg = await uart.read(100)
    # print('[DEBUG] read', msg)
    unescaped = unescape(msg)
    sof = unescaped[0]
    adr = unescaped[1]
    cmd = unescaped[2]
    state = unescaped[3]
    length = unescaped[4]
    rxdata = unescaped[5:-2]
    chk = unescaped[-2]
    eof = unescaped[-1]
    assert (sof, adr, eof) == (0x7e, 0x00, 0x7e)
    assert state == 0x00
    assert length == len(rxdata)
    frame = bytes([adr, cmd, state, length]) + rxdata
    assert checksum(frame) == chk
    return cmd, rxdata


START_MEASUREMENT = 0x00
STOP_MEASUREMENT = 0x01
READ_MEASURED_VALUES = 0x03
READ_WRITE_AUTO_CLEANING_INTERVAL = 0x80
START_FAN_CLEANING = 0x56
DEVICE_RESET = 0xd3


async def read_meas(uart: asyncio.Stream) -> tuple[float, float, float, float]:
    await mosi(uart, READ_MEASURED_VALUES, [])
    _, rxdata = await miso(uart)
    if not rxdata:
        print('[INFO] No new measurements are available')
        return (None, None, None, None)
    # print('[INFO] READ_MEASURED_VALUES:', rxdata)
    pm1 = struct.unpack('>f', rxdata[0:4])[0]
    pm25 = struct.unpack('>f', rxdata[4:8])[0]
    pm4 = struct.unpack('>f', rxdata[8:12])[0]
    pm10 = struct.unpack('>f', rxdata[12:16])[0]
    print(f'[INFO] PM1.0: {pm1}, PM2.5: {pm25}, PM4.0: {pm4}, PM10: {pm10}')
    return (pm1, pm25, pm4, pm10)


async def start_http_server():
    ser = machine.UART(1,
                       baudrate=115200,
                       rx=0,
                       tx=1,
                       timeout_char=50,
                       timeout=50)
    stream = asyncio.StreamReader(ser)

    await mosi(stream, DEVICE_RESET, [])
    print(f'[INFO] DEVICE_RESET: {await miso(stream)}')

    await mosi(stream, START_MEASUREMENT, [0x01, 0x03])
    print(f'[INFO] START_MEASUREMENT: {await miso(stream)}')

    addr = socket.getaddrinfo('0.0.0.0', 9318)[0][-1]
    sock = socket.socket()
    sock.bind(addr)
    sock.listen(1)
    print('[INFO] Listening on', addr)

    while True:
        try:
            await handle_client(sock, stream)
            # await read_meas(stream)
            # await asyncio.sleep(1)
        except Exception as e:
            print(e)


async def handle_client(sock, ser):
    cl, addr = sock.accept()
    print('[INFO] Client connected from', addr)
    try:
        cl_file = cl.makefile('rwb', 0)
        while True:
            line = cl_file.readline()
            if not line or line == b'\r\n':
                break
        pm1, pm25, pm4, pm10 = await read_meas(ser)
        response = f'''# HELP air_pm1 PM 1.0 (ug/m3)
# TYPE air_pm1 gauge
air_pm1 {pm1}
# HELP air_pm25 PM 2.5 (ug/m3)
# TYPE air_pm25 gauge
air_pm25 {pm25}
# HELP air_pm4 PM 4.0 (ug/m3)
# TYPE air_pm4 gauge
air_pm4 {pm4}
# HELP air_pm10 PM 10 (ug/m3)
# TYPE air_pm10 gauge
air_pm10 {pm10}
'''
        cl.send(f'''HTTP/1.0 200 OK\r
Content-type: text/plain; version=0.0.4\r
Content-Length: {len(response)}\r
\r
{response}''')
    finally:
        cl_file.close()
        cl.close()