scan_ble

#!/usr/bin/python3
#
# (C) 2022 Arnaud Patard <arnaud.patard@rtp-net.org>
#    This program is free software; you can redistribute it and/or modify
#    it under the terms of the GNU General Public License as published by
#    the Free Software Foundation; either version 3 of the License, or
#    (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    You should have received a copy of the GNU General Public License
#    along with this program; if not, please see
#    <http://www.gnu.org/licenses/>

import asyncio
import logging
import sys
import time
from struct import unpack
import requests
import binascii
import yaml
import argparse
from typing import Any

from bleak import BleakScanner
from bleak.assigned_numbers import AdvertisementDataType
from bleak.backends.bluezdbus.advertisement_monitor import OrPattern
from bleak.backends.bluezdbus.scanner import BlueZScannerArgs
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData


default_cfg = {
    "adapter": "hci0",
    # seconds
    "sleep": 1,
    # seconds
    "ratelimit": 300,
    # not used yet.
    "loglevel": "warning",
    "domoticz": {"hostname": "localhost:8080", "ssl": True},
    "sensors": [],
}
cfg = {}
request_timestamp = None
logger = logging.getLogger(__name__)
sensor_list = {}
loglevels = {
    "debug": logging.DEBUG,
    "info": logging.INFO,
    "warning": logging.WARNING,
    "error": logging.ERROR,
    "critical": logging.CRITICAL,
}


class Sensor(object):
    def __new__(cls, data):
        if cls == Sensor:
            for sc in Sensor.__subclasses__():
                try:
                    return sc(data)
                except ValueError:
                    pass
            raise ValueError("Can't find valid decoder class")
        return super().__new__(cls)

    def get_humidity_status(self, hum):
        if hum < 40:
            return 2
        elif hum >= 40 and hum < 70:
            return 1
        elif hum >= 70:
            return 3
        else:
            return 0


class XiaomiTempHum(Sensor):
    def __init__(self, data):
        if data.local_name != "MJ_HT_V1":
            raise ValueError("Not compatible")
        self.battery = None
        self.temp = None
        self.hum = None
        self.hum_stat = None

    def decode(self, dev, data):
        svc_data = data.service_data
        valid_data = False
        # str = None
        for k, v in svc_data.items():
            if k.startswith("0000fe95-"):
                data_len = int(v[13])
                data_str = binascii.hexlify(v[14 : 14 + data_len + 11])
                logger.debug(f"{dev.address}: {v[11]:02x} {v[13]:02x} {data_str}")
                if v[11] == 0x0D:
                    self.temp = unpack("h", v[14:16])[0] / 10
                    self.hum = unpack("h", v[16:18])[0] / 10
                    self.hum_stat = self.get_humidity_status(self.hum)
                    # str = f"&nvalue=0&svalue={self.temp};{self.hum};{self.hum_stat}"
                    valid_data = True
                elif v[11] == 0x04:
                    self.temp = unpack("h", v[14:16])[0] / 10
                    # str = f"&nvalue=0&svalue={self.temp}"
                    valid_data = True
                elif v[11] == 0x06:
                    self.hum = unpack("h", v[14:16])[0] / 10
                    self.hum_stat = self.get_humidity_status(self.hum)
                    # str = f"&nvalue={self.hum}&svalue={self.hum_stat}"
                    valid_data = True
                elif v[11] == 0x0A:
                    self.battery = f"{v[14]}"
                    # str = ""
                    valid_data = True
                else:
                    raise ValueError(f"Unknown type {v[11]}")
                break
        # if str is None:
        if not valid_data:
            raise ValueError(
                f"{dev.address}: Unable to find valid information in service data"
            )
        # dont know how to translate dbm to domoticz values
        # str = f"{str}&rssi={dev.rssi}"
        str = ""
        if self.battery is not None and self.temp is not None and self.hum is not None:
            # str = f"{str}&battery={self.battery}"
            str = f"&nvalue=0&svalue={self.temp};{self.hum};{self.hum_stat}&battery={self.battery}"  # noqa 501
        return str


def parse_data(sensor_list: dict[str, Any], dev: BLEDevice, d: AdvertisementData):
    global request_timestamp
    global cfg
    mac_addr = dev.address
    svc_data = d
    idx = None
    if mac_addr not in sensor_list:
        try:
            s = Sensor(svc_data)
        except ValueError:
            logger.info(f"Sensor {mac_addr} ({svc_data.local_name}): type not found")
            return

        sensor_list[mac_addr] = {"class": s}
        for i in cfg["sensors"]:
            if mac_addr == i["address"]:
                sensor_list[mac_addr]["idx"] = i["idx"]
                idx = i["idx"]
    else:
        s = sensor_list[mac_addr]["class"]
        if "idx" in sensor_list[mac_addr]:
            idx = sensor_list[mac_addr]["idx"]
    try:
        str = s.decode(dev, svc_data)
    except ValueError as e:
        logger.warning(f"Sensor {mac_addr}: {e}")
        return
    if str == "":
        return
    if idx is not None:
        if request_timestamp is not None and int(time.time()) < (
            request_timestamp + cfg["ratelimit"]
        ):
            return
        uri_scheme = "https"
        if "ssl" in cfg["domoticz"] and cfg["domoticz"]["ssl"] is False:
            uri_scheme = "http"
        url = f"{uri_scheme}://{cfg['domoticz']['hostname']}/json.htm?type=command&param=udevice&idx={idx}{str}"  # noqa 501
        logger.info(f"Calling {url}")
        try:
            if "login" in cfg["domoticz"] and "password" in cfg["domoticz"]:
                r = requests.get(
                    url, auth=(cfg["domoticz"]["login"], cfg["domoticz"]["password"])
                )
            else:
                r = requests.get(url)
            if r.status_code != 200:
                logger.error(
                    f"HTTP error {r.status_code} while posting data to domoticz"
                )
                return
        except Exception as e:
            logger.error(f"Error {e} while posting data to domoticz")
            return

        json_data = r.json()
        logger.info(f"Domoticz reply: {json_data}")
        if "status" not in json_data or json_data["status"] != "OK":
            logger.error(f"Failed to update sensor {mac_addr} index {idx}")
        request_timestamp = int(time.time())
    else:
        logger.debug(f"No index for sensor {mac_addr}. Not in configuration ?")


async def simple_callback(device: BLEDevice, advertisement_data: AdvertisementData):
    if advertisement_data.local_name is not None:
        parse_data(sensor_list, device, advertisement_data)
    else:
        # put device in bluez cache and get local name
        await BleakScanner.find_device_by_address(device.address)


async def main():
    scanner = BleakScanner(
        simple_callback,
        scanning_mode="passive",
        bluez=BlueZScannerArgs(
            or_patterns=[
                OrPattern(0, AdvertisementDataType.FLAGS, b"\x06"),
                OrPattern(0, AdvertisementDataType.FLAGS, b"\x1a"),
            ]
        ),
        adapter=cfg["adapter"],
    )

    while True:
        await scanner.start()
        await asyncio.sleep(cfg["sleep"])
        await scanner.stop()


def cfg_loglevel_translate(loglevel):
    if loglevel not in loglevels:
        return logging.WARNING
    return loglevels[loglevel]


def parse_cfg(cfgfile):
    cfg_data = {}
    with open(cfgfile, "r") as fd:
        try:
            cfg_data = yaml.safe_load(fd)
        except yaml.YAMLError as e:
            print(e)
            sys.exit(1)
    cfg = default_cfg
    cfg.update(cfg_data)
    return cfg


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Parse BLE advertisement and send data to domoticz"
    )
    parser.add_argument(
        "--cfg",
        dest="cfgfile",
        action="store",
        default="/etc/scan_ble.yml",
        help="configuration file",
    )
    args = parser.parse_args()
    cfg = parse_cfg(args.cfgfile)

    logging.basicConfig(
        level=cfg_loglevel_translate(cfg["loglevel"]),
        format="%(asctime)-15s %(name)-8s %(levelname)s: %(message)s",
    )
    asyncio.run(main())