diff --git a/example.py b/example.py index c5e2b42a..17f19239 100644 --- a/example.py +++ b/example.py @@ -6,6 +6,7 @@ import asyncio import json import logging +import os import signal import sys import traceback @@ -206,14 +207,20 @@ def print_rpc_device(device: RpcDevice) -> None: def save_endpoints(device: BlockDevice | RpcDevice) -> None: """Save device endpoints.""" - data_raw = {} - data_raw.update({"config": device.config}) - data_raw.update({"shelly": device.shelly}) - data_raw.update({"status": device.status}) + data_raw = {"shelly": device.shelly, "status": device.status} - data_normalized = normalize_device_data(data_raw) + if device.gen in BLOCK_GENERATIONS: + data_raw.update({"settings": device.settings}) + data_normalized = _normalize_block_data(data_raw) + else: + data_raw.update({"config": device.config}) + data_normalized = _normalize_rpc_data(data_raw) + + folder = "fixtures" + if not os.path.exists(folder): + os.makedirs(folder) with open( - f"fixtures/{device.model}- {MODEL_NAMES[device.model]}.json", + f"{folder}/gen{device.gen} - {device.model} - {MODEL_NAMES[device.model]}.json", "w", encoding="utf-8", ) as file: @@ -222,85 +229,140 @@ def save_endpoints(device: BlockDevice | RpcDevice) -> None: sys.exit(0) -def normalize_device_data(data: dict[str, Any]) -> dict[str, Any]: - """Normalize device data for testing.""" +def _replacement_values() -> dict[str, Any]: + return { + "wifi": "Wifi-Network-Name", + "wifi_mac": "11:22:33:44:55:66", + "device_mac": "AABBCCDDEEFF", + "device_name": "Test Name", + "switch_name": "Switch Test Name", + "input_name": "Input Test Name", + "em_name": "Energy Monitor Test Name", + } + + +def _normalize_block_data(data: dict[str, Any]) -> dict[str, Any]: + """Normalize data for BLOCK devices.""" + status = data["status"] + shelly = data["shelly"] + settings = data["settings"] + values = _replacement_values() + + real_mac: str = status["mac"] + + # Shelly endpoint + shelly["name"] = values["device_name"] + shelly["mac"] = values["device_mac"] + + # Status endpoint + status["mac"] = values["device_mac"] + + if status["wifi_sta"].get("ssid"): + status["wifi_sta"]["ssid"] = values["wifi"] + + # Config endpoint + settings["device"]["hostname"] = settings["device"]["hostname"].replace( + real_mac, values["device_mac"] + ) + settings["device"]["mac"] = values["device_mac"] + + # Some devices use MAC uppercase, others lowercase + settings["mqtt"]["id"] = settings["mqtt"]["id"].replace( + real_mac, values["device_mac"] + ) + settings["mqtt"]["id"] = settings["mqtt"]["id"].replace( + real_mac.lower(), values["device_mac"].lower() + ) + + settings["name"] = values["device_name"] + settings["wifi_ap"]["ssid"] = settings["wifi_ap"]["ssid"].replace( + real_mac, values["device_mac"] + ) + settings["wifi_sta"]["ssid"] = values["wifi"] + + # Updata dataset + data.update({"settings": settings}) + data.update({"shelly": shelly}) + data.update({"status": status}) + + return data + + +def _normalize_rpc_data(data: dict[str, Any]) -> dict[str, Any]: + """Normalize data for RPC devices.""" config = data["config"] status = data["status"] shelly = data["shelly"] real_mac: str = status["sys"]["mac"] - wifi = "Wifi-Network-Name" - wifi_mac = "11:22:33:44:55:66" - device_mac = "AABBCCDDEEFF" - device_name = "Test Name" - switch_name = "Switch Test Name" - input_name = "Input Test Name" - em_name = "Energy Monitor Test Name" + values = _replacement_values() # Config endpoint - config["sys"]["device"]["name"] = device_name - config["sys"]["device"]["mac"] = device_mac + config["sys"]["device"]["name"] = values["device_name"] + config["sys"]["device"]["mac"] = values["device_mac"] # Some devices use MAC uppercase, others lowercase config["mqtt"]["client_id"] = config["mqtt"]["client_id"].replace( - real_mac, device_mac + real_mac, values["device_mac"] ) config["mqtt"]["client_id"] = config["mqtt"]["client_id"].replace( - real_mac.lower(), device_mac.lower() + real_mac.lower(), values["device_mac"].lower() ) if config["mqtt"].get("topic_prefix"): # Some devices use MAC uppercase, others lowercase config["mqtt"]["topic_prefix"] = config["mqtt"]["topic_prefix"].replace( - real_mac, device_mac + real_mac, values["device_mac"] ) config["mqtt"]["topic_prefix"] = config["mqtt"]["topic_prefix"].replace( - real_mac.lower(), device_mac.lower() + real_mac.lower(), values["device_mac"].lower() ) for id_ in range(5): if config.get(f"thermostat:{id_}"): config[f"thermostat:{id_}"]["sensor"] = config[f"thermostat:{id_}"][ "sensor" - ].replace(real_mac.lower(), device_mac.lower()) + ].replace(real_mac.lower(), values["device_mac"].lower()) config[f"thermostat:{id_}"]["actuator"] = config[f"thermostat:{id_}"][ "actuator" - ].replace(real_mac.lower(), device_mac.lower()) + ].replace(real_mac.lower(), values["device_mac"].lower()) if config.get(f"switch:{id_}"): - config[f"switch:{id_}"]["name"] = f"{switch_name} {id_}" + config[f"switch:{id_}"]["name"] = f"{values['switch_name']} {id_}" if config.get(f"input:{id_}"): - config[f"input:{id_}"]["name"] = f"{input_name} {id_}" + config[f"input:{id_}"]["name"] = f"{values['input_name']} {id_}" if config.get(f"em:{id_}"): - config[f"em:{id_}"]["name"] = f"{em_name} {id_}" + config[f"em:{id_}"]["name"] = f"{values['em_name']} {id_}" if config.get("wifi"): - config["wifi"] = wifi + config["wifi"] = values["wifi"] # Shelly endpoint - shelly["name"] = device_name + shelly["name"] = values["device_name"] # Some devices use MAC uppercase, others lowercase - shelly["id"] = shelly["id"].replace(real_mac, device_mac) - shelly["id"] = shelly["id"].replace(real_mac.lower(), device_mac.lower()) - shelly["mac"] = device_mac + shelly["id"] = shelly["id"].replace(real_mac, values["device_mac"]) + shelly["id"] = shelly["id"].replace(real_mac.lower(), values["device_mac"].lower()) + shelly["mac"] = values["device_mac"] # Status endpoint - status["sys"]["mac"] = device_mac + status["sys"]["mac"] = values["device_mac"] if status["wifi"].get("ssid"): - status["wifi"]["ssid"] = wifi + status["wifi"]["ssid"] = values["wifi"] if status["wifi"].get("mac"): - status["wifi"]["mac"] = wifi_mac + status["wifi"]["mac"] = values["wifi_mac"] if status["sys"].get("id"): # Some devices use MAC uppercase, others lowercase - status["sys"]["id"] = status["sys"]["id"].replace(real_mac, device_mac) status["sys"]["id"] = status["sys"]["id"].replace( - real_mac.lower(), device_mac.lower() + real_mac, values["device_mac"] + ) + status["sys"]["id"] = status["sys"]["id"].replace( + real_mac.lower(), values["device_mac"].lower() ) # Updata dataset