Strict typing

This commit is contained in:
Scott Wallace 2021-12-11 16:13:22 +00:00
parent a921c3b88f
commit cf4a2a954b
Signed by: scott
GPG key ID: AA742FDC5AFE2A72

View file

@ -4,11 +4,12 @@ Read Xiaomi LYWSD03MMC advertised packets and send them to MQTT
import os import os
from datetime import datetime from datetime import datetime
from socket import gaierror from socket import gaierror
from typing import TypedDict
import bluetooth._bluetooth as bluez import bluetooth._bluetooth as bluez # type: ignore[import]
import paho.mqtt.publish as publish import paho.mqtt.publish as publish # type: ignore[import]
from bluetooth_utils.bluetooth_utils import ( from bluetooth_utils.bluetooth_utils import ( # type: ignore[import]
disable_le_scan, disable_le_scan,
enable_le_scan, enable_le_scan,
parse_le_advertising_events, parse_le_advertising_events,
@ -17,9 +18,27 @@ from bluetooth_utils.bluetooth_utils import (
) )
def send_to_mqtt(data: dict): class MQTTdata(TypedDict):
"""
Class to define the MQTT data to send
Args:
TypedDict ([type]): timestamp, MAC addr, temp, humidity %, battery %
"""
timestamp: str
mac: str
temperature: float
humidity: int
battery: int
def send_to_mqtt(data: MQTTdata) -> None:
""" """
Send data from LYWSD03MMC to MQTT Send data from LYWSD03MMC to MQTT
Args:
data (MQTTdata): [description]
""" """
msgs = [ msgs = [
{ {
@ -38,22 +57,29 @@ def send_to_mqtt(data: dict):
publish.multiple(msgs, hostname=os.environ.get('MQTT_HOST', 'mqtt')) publish.multiple(msgs, hostname=os.environ.get('MQTT_HOST', 'mqtt'))
def le_advertise_packet_handler( def le_advertise_packet_handler( # pylint: disable=unused-argument
mac: str, mac: str,
adv_type: int, # pylint: disable=unused-argument adv_type: int,
data: bytes, data: bytes,
rssi: int, # pylint: disable=unused-argument rssi: int,
): ) -> None:
""" """
Handle new Xiaomi LYWSD03MMC BTLE advertise packet Handle new Xiaomi LYWSD03MMC BTLE advertise packet
Args:
mac (str): MAC address of the sensor
adv_type (int): NOT USED
data (bytes): data from sensor
rssi (int): NOT USED
""" """
data_str = raw_packet_to_str(data) data_str = raw_packet_to_str(data)
# Check for ATC preamble # Check for ATC preamble
if data_str[0:2] == '11' and data_str[6:10] == '1a18': if data_str[0:2] == '11' and data_str[6:10] == '1a18':
temp = int(data_str[22:26], 16) / 10 temp = int(data_str[22:26], 16) / 10
hum = int(data_str[26:28], 16) hum = int(data_str[26:28], 16)
batt = int(data_str[28:30], 16) batt = int(data_str[28:30], 16)
mqtt_data = { mqtt_data: MQTTdata = {
'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
'mac': mac, 'mac': mac,
'temperature': temp, 'temperature': temp,
@ -68,7 +94,7 @@ def le_advertise_packet_handler(
if __name__ == '__main__': if __name__ == '__main__':
def main(): def main() -> None:
""" """
Main program Main program
""" """