#!/usr/bin/env python3
"""
Synchronizuje friendly_names i dashboard IBSystem z dokumentacji projektu.
Źródło: /smart-home/proj-ib-lighting-rs485-zarki.yaml
"""
import yaml, re, sys, os, subprocess, shutil, io
from datetime import datetime
import copy
PROJECT_YAML = "/smart-home/proj-ib-lighting-rs485-zarki.yaml"
CONFIG_YAML = "/ibsystem/ibsystem2mqtt.yaml"
DASHBOARD_LOCAL = "/ibsystem/ibsystem_dashboard.yaml"
DASHBOARD_HA = "root@192.168.50.151:/config/dashboards/ibsystem_dashboard.yaml"
HA_SSH_PORT = 2222
MQTT_HOST = "192.168.50.151"
MQTT_USER = "mqtt"
MQTT_PASS = "mqtt123"
# Zakresy ID sterowników oświetlenia (pomijamy ogrzewanie itp.)
LIGHTING_ID_MIN = 30
LIGHTING_ID_MAX = 99
def log(msg): print(f"[{datetime.now():%H:%M:%S}] {msg}")
def device_id_to_rs_id(device_id):
m = re.match(r"rs\.(\d+)\.id\.(\d+)", device_id)
return (m.group(1), m.group(2)) if m else (None, None)
def io_key(rs, dev_id, io_id):
safe = re.sub(r"[^a-zA-Z0-9]+", "_", io_id).strip("_")
return f"rs{rs}_id{dev_id}_{safe}"
def entity_id(rs, dev_id, io_id):
safe = re.sub(r"[^a-zA-Z0-9]+", "_", io_id).strip("_")
return f"switch.ibsystem_rs{rs}_id{dev_id}_rs{rs}_id{dev_id}_{safe}"
def parse_project_yaml(path):
with open(path) as f:
data = yaml.safe_load(f)
devices = data["project"]["hardware_specification"]["devices"]
result = []
for dev in devices:
device_id = dev.get("device_id", "")
location = dev.get("location", "").strip()
rs, dev_id = device_id_to_rs_id(device_id)
if not rs:
continue
io_list = []
for io in dev.get("io", []):
io_id = io.get("io_id", "")
desc = io.get("description", "").strip()
status = io.get("status", "ok")
if io_id.startswith("output.do.") and desc and status != "nc" \
and not desc.startswith("Opcjonalne"):
io_list.append({"io_id": io_id, "description": desc})
result.append({
"device_id": device_id,
"rs": rs,
"dev_id": dev_id,
"location": location,
"io": io_list,
})
return result
def build_friendly_names(devices):
names = {}
for dev in devices:
rs, dev_id, location = dev["rs"], dev["dev_id"], dev["location"]
if location:
names[f"rs{rs}_id{dev_id}"] = f"{location} - ID{dev_id}"
for io in dev["io"]:
names[io_key(rs, dev_id, io["io_id"])] = io["description"]
return names
def build_dashboard(devices):
# Tylko sterowniki oświetlenia (ID30+)
light_devs = [d for d in devices
if d["io"] and LIGHTING_ID_MIN <= int(d["dev_id"]) <= LIGHTING_ID_MAX]
# Lista wszystkich switch entity_id dla WSZYSTKIE ON/OFF
all_switches = []
for dev in light_devs:
rs, dev_id = dev["rs"], dev["dev_id"]
for io in dev["io"]:
all_switches.append(entity_id(rs, dev_id, io["io_id"]))
# Karty per urządzenie
device_cards = []
for dev in light_devs:
rs, dev_id, location = dev["rs"], dev["dev_id"], dev["location"]
title = f"{location} (ID{dev_id})" if location else f"ID{dev_id}"
entities = []
for io in dev["io"]:
entities.append({
"entity": entity_id(rs, dev_id, io["io_id"]),
"name": io["description"],
})
device_cards.append({
"type": "entities",
"title": title,
"entities": entities,
})
dashboard = {
"title": "IBSystem Sterowniki",
"views": [{
"title": "Wszystkie sterowniki",
"path": "all",
"icon": "mdi:lightbulb-group",
"cards": [
{
"type": "horizontal-stack",
"cards": [
{
"type": "button",
"name": "WSZYSTKIE ON",
"icon": "mdi:lightbulb-group",
"icon_height": "40px",
"tap_action": {
"action": "call-service",
"service": "switch.turn_on",
"target": {"entity_id": copy.deepcopy(all_switches)},
},
},
{
"type": "button",
"name": "WSZYSTKIE OFF",
"icon": "mdi:lightbulb-group-off",
"icon_height": "40px",
"tap_action": {
"action": "call-service",
"service": "switch.turn_off",
"target": {"entity_id": copy.deepcopy(all_switches)},
},
},
],
},
] + device_cards,
}],
}
return dashboard
def update_friendly_names(config_path, new_names, dry_run=False):
with open(config_path) as f:
config = yaml.safe_load(f)
old_names = config.get("friendly_names") or {}
changed = {k for k in (set(old_names) | set(new_names)) if old_names.get(k) != new_names.get(k)}
if not changed:
log("friendly_names: bez zmian.")
return False
log(f"friendly_names: {len(changed)} zmian.")
if dry_run:
for k in sorted(changed):
print(f" {k}: {old_names.get(k)!r} → {new_names.get(k)!r}")
return True
shutil.copy(config_path, config_path + ".bak")
config["friendly_names"] = new_names
with open(config_path, "w") as f:
yaml.dump(config, f, allow_unicode=True, default_flow_style=False, sort_keys=False)
log(f" Zapisano {config_path}")
return True
def update_dashboard(dashboard, local_path, ha_path, dry_run=False):
# Serializuj do YAML
buf = io.StringIO()
yaml.dump(dashboard, buf, allow_unicode=True, default_flow_style=False,
sort_keys=False, width=200)
new_content = buf.getvalue()
# Sprawdź czy się zmieniło
try:
with open(local_path) as f:
old_content = f.read()
if old_content == new_content:
log("dashboard: bez zmian.")
return False
except FileNotFoundError:
pass
log("dashboard: aktualizuję.")
if dry_run:
log(" (dry-run, nie zapisuję)")
return True
with open(local_path, "w") as f:
f.write(new_content)
# Kopiuj na HA przez SSH
host, remote_path = ha_path.split(":", 1)
result = subprocess.run(
["scp", "-P", str(HA_SSH_PORT), "-o", "StrictHostKeyChecking=no",
local_path, ha_path],
capture_output=True, text=True
)
if result.returncode != 0:
log(f" BŁĄD SCP: {result.stderr.strip()}")
else:
log(f" Skopiowano → {ha_path}")
return True
def reload_ibsystem():
log("Czyszczę MQTT discovery cache...")
os.system(
f"timeout 5 mosquitto_sub -h {MQTT_HOST} -u {MQTT_USER} -P {MQTT_PASS} "
f"-t 'homeassistant/#' --retained-only -v 2>/dev/null | "
f"grep '/config' | grep 'ibsystem' | awk '{{print $1}}' | "
f"while read t; do mosquitto_pub -h {MQTT_HOST} -u {MQTT_USER} -P {MQTT_PASS} "
f"-t \"$t\" -n -r; done"
)
log("Restartuję ibsystem2mqtt...")
subprocess.run(["systemctl", "restart", "ibsystem2mqtt"], check=True)
def main():
dry_run = "--dry-run" in sys.argv
log(f"Parsing {PROJECT_YAML}...")
try:
devices = parse_project_yaml(PROJECT_YAML)
except Exception as e:
log(f"BŁĄD parsowania: {e}")
sys.exit(1)
log(f" {len(devices)} urządzeń, "
f"{sum(len(d['io']) for d in devices)} wyjść DO.")
names_changed = update_friendly_names(CONFIG_YAML, build_friendly_names(devices), dry_run)
dashboard_changed = update_dashboard(build_dashboard(devices), DASHBOARD_LOCAL,
DASHBOARD_HA, dry_run)
if (names_changed or dashboard_changed) and not dry_run:
reload_ibsystem()
else:
log("Brak zmian wymagających restartu.")
if __name__ == "__main__":
main()
