791 lines
25 KiB
Python
791 lines
25 KiB
Python
import asyncio
|
|
import ipaddress
|
|
import socket
|
|
from pathlib import Path
|
|
|
|
import aiohttp
|
|
|
|
from . import paths
|
|
from .profiles import load_yaml, now_iso, save_yaml, slugify
|
|
|
|
PROBE_TIMEOUT = 1.5
|
|
CONTROL_TIMEOUT = 6.0
|
|
SCAN_CONCURRENCY = 64
|
|
|
|
PROFILE_HEADER = """
|
|
Shelly device profile.
|
|
|
|
Generated by: solixauto discover-shelly
|
|
This device is an ACTUATOR. The automation engine turns the listed
|
|
channels on and off over local HTTP. No cloud service is involved.
|
|
|
|
channels: switch outputs available on this device.
|
|
readable: status fields observed at generation time.
|
|
auth: set username/password here if the device requires it.
|
|
|
|
If the device gets a new IP, either give it a DHCP reservation or set
|
|
host: to its mDNS name and rerun discovery.
|
|
"""
|
|
|
|
|
|
def local_addresses():
|
|
found = []
|
|
|
|
probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
probe.connect(("8.8.8.8", 80))
|
|
found.append(probe.getsockname()[0])
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
probe.close()
|
|
|
|
try:
|
|
for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
|
|
found.append(info[4][0])
|
|
except socket.gaierror:
|
|
pass
|
|
|
|
try:
|
|
import ifaddr
|
|
|
|
for adapter in ifaddr.get_adapters():
|
|
for ip in adapter.ips:
|
|
if ip.is_IPv4:
|
|
found.append(ip.ip)
|
|
except Exception:
|
|
pass
|
|
|
|
usable = []
|
|
for address in found:
|
|
try:
|
|
parsed = ipaddress.ip_address(address)
|
|
except ValueError:
|
|
continue
|
|
if parsed.is_loopback or parsed.is_link_local or not parsed.is_private:
|
|
continue
|
|
if address not in usable:
|
|
usable.append(address)
|
|
|
|
return usable
|
|
|
|
|
|
def local_subnets():
|
|
networks = []
|
|
for address in local_addresses():
|
|
network = ipaddress.ip_network(f"{address}/24", strict=False)
|
|
if network not in networks:
|
|
networks.append(network)
|
|
return networks
|
|
|
|
|
|
def local_subnet():
|
|
networks = local_subnets()
|
|
return networks[0] if networks else None
|
|
|
|
|
|
def auth_from(profile):
|
|
auth = (profile or {}).get("auth") or {}
|
|
username = auth.get("username")
|
|
password = auth.get("password")
|
|
if username and password:
|
|
return aiohttp.BasicAuth(username, password)
|
|
return None
|
|
|
|
|
|
async def probe_host(session, host, auth=None):
|
|
url = f"http://{host}/shelly"
|
|
try:
|
|
async with session.get(
|
|
url, timeout=aiohttp.ClientTimeout(total=PROBE_TIMEOUT), auth=auth
|
|
) as response:
|
|
if response.status != 200:
|
|
return None
|
|
payload = await response.json(content_type=None)
|
|
except Exception:
|
|
return None
|
|
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
if not ({"mac", "type", "id", "model"} & set(payload)):
|
|
return None
|
|
|
|
payload["_host"] = str(host)
|
|
payload["_gen"] = int(payload.get("gen", 1) or 1)
|
|
return payload
|
|
|
|
|
|
async def scan_network(network, verbose=True):
|
|
hosts = list(network.hosts())
|
|
if verbose:
|
|
print(f"Scanning {network} ({len(hosts)} addresses)...")
|
|
|
|
found = []
|
|
semaphore = asyncio.Semaphore(SCAN_CONCURRENCY)
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
|
|
async def worker(host):
|
|
async with semaphore:
|
|
result = await probe_host(session, host)
|
|
if result:
|
|
found.append(result)
|
|
if verbose:
|
|
print(f" found {result['_host']} ({describe(result)})")
|
|
|
|
await asyncio.gather(*(worker(host) for host in hosts))
|
|
|
|
return found
|
|
|
|
|
|
async def probe_hosts(hosts, verbose=True):
|
|
found = []
|
|
async with aiohttp.ClientSession() as session:
|
|
for host in hosts:
|
|
result = await probe_host(session, host)
|
|
if result:
|
|
found.append(result)
|
|
if verbose:
|
|
print(f" found {host} ({describe(result)})")
|
|
elif verbose:
|
|
print(f" no Shelly at {host}")
|
|
return found
|
|
|
|
|
|
def mdns_discover(seconds=5, verbose=True):
|
|
try:
|
|
from zeroconf import ServiceBrowser, Zeroconf
|
|
except ImportError:
|
|
if verbose:
|
|
print("zeroconf not installed, skipping mDNS (pip install zeroconf)")
|
|
return []
|
|
|
|
import time
|
|
|
|
discovered = []
|
|
|
|
class Listener:
|
|
def add_service(self, zeroconf_instance, service_type, name):
|
|
info = zeroconf_instance.get_service_info(service_type, name, timeout=2000)
|
|
if not info:
|
|
return
|
|
for raw in info.parsed_addresses():
|
|
discovered.append(raw)
|
|
|
|
def update_service(self, *args):
|
|
pass
|
|
|
|
def remove_service(self, *args):
|
|
pass
|
|
|
|
zeroconf_instance = Zeroconf()
|
|
listener = Listener()
|
|
browsers = [
|
|
ServiceBrowser(zeroconf_instance, "_shelly._tcp.local.", listener),
|
|
ServiceBrowser(zeroconf_instance, "_http._tcp.local.", listener),
|
|
]
|
|
if verbose:
|
|
print(f"Listening for mDNS announcements for {seconds}s...")
|
|
time.sleep(seconds)
|
|
for browser in browsers:
|
|
browser.cancel()
|
|
zeroconf_instance.close()
|
|
|
|
return sorted(set(discovered))
|
|
|
|
|
|
def describe(payload):
|
|
gen = payload.get("_gen", 1)
|
|
if gen >= 2:
|
|
return f"{payload.get('app') or payload.get('model')} gen{gen}"
|
|
return f"{payload.get('type')} gen1"
|
|
|
|
|
|
async def fetch_config(session, host, gen, auth=None):
|
|
path = "/rpc/Shelly.GetConfig" if gen >= 2 else "/settings"
|
|
url = f"http://{host}{path}"
|
|
try:
|
|
async with session.get(
|
|
url, timeout=aiohttp.ClientTimeout(total=CONTROL_TIMEOUT), auth=auth
|
|
) as response:
|
|
if response.status != 200:
|
|
return {}
|
|
return await response.json(content_type=None)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def device_name_from(payload, config, gen):
|
|
candidates = []
|
|
|
|
if gen >= 2:
|
|
system = (config or {}).get("sys") or {}
|
|
device = system.get("device") or {}
|
|
candidates.append(device.get("name"))
|
|
else:
|
|
candidates.append((config or {}).get("name"))
|
|
settings_device = (config or {}).get("device") or {}
|
|
candidates.append(settings_device.get("hostname"))
|
|
|
|
candidates.append((payload or {}).get("name"))
|
|
|
|
for candidate in candidates:
|
|
if candidate and str(candidate).strip():
|
|
return str(candidate).strip()
|
|
return ""
|
|
|
|
|
|
def channel_name_from(config, gen, index):
|
|
if not config:
|
|
return ""
|
|
if gen >= 2:
|
|
entry = config.get(f"switch:{index}") or {}
|
|
name = entry.get("name")
|
|
else:
|
|
relays = config.get("relays") or []
|
|
name = relays[index].get("name") if index < len(relays) else None
|
|
return str(name).strip() if name else ""
|
|
|
|
|
|
async def fetch_status(session, host, gen, auth=None):
|
|
path = "/rpc/Shelly.GetStatus" if gen >= 2 else "/status"
|
|
url = f"http://{host}{path}"
|
|
try:
|
|
async with session.get(
|
|
url, timeout=aiohttp.ClientTimeout(total=CONTROL_TIMEOUT), auth=auth
|
|
) as response:
|
|
if response.status != 200:
|
|
return {}
|
|
return await response.json(content_type=None)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
DAY_NAMES = {
|
|
"0": "Sun", "1": "Mon", "2": "Tue", "3": "Wed",
|
|
"4": "Thu", "5": "Fri", "6": "Sat", "7": "Sun",
|
|
"SUN": "Sun", "MON": "Mon", "TUE": "Tue", "WED": "Wed",
|
|
"THU": "Thu", "FRI": "Fri", "SAT": "Sat",
|
|
}
|
|
|
|
|
|
def describe_cron(spec):
|
|
parts = str(spec or "").split()
|
|
if len(parts) != 6:
|
|
return str(spec)
|
|
|
|
second, minute, hour, day, month, weekday = parts
|
|
|
|
if not (second.isdigit() and minute.isdigit() and hour.isdigit()):
|
|
return str(spec)
|
|
|
|
clock = f"{int(hour):02d}:{int(minute):02d}"
|
|
|
|
if weekday in ("*", "?") and day in ("*", "?"):
|
|
return f"daily at {clock}"
|
|
|
|
if weekday not in ("*", "?"):
|
|
names = []
|
|
for token in weekday.replace("-", ",").split(","):
|
|
token = token.strip().upper()
|
|
names.append(DAY_NAMES.get(token, token))
|
|
|
|
unique = []
|
|
for name in names:
|
|
if name not in unique:
|
|
unique.append(name)
|
|
|
|
if len(unique) == 7:
|
|
return f"daily at {clock}"
|
|
|
|
return f"{clock} on {', '.join(unique)}"
|
|
|
|
return f"{clock} (day {day}, month {month})"
|
|
|
|
|
|
def describe_job(job):
|
|
timespec = job.get("timespec") or job.get("cron") or ""
|
|
when = describe_cron(timespec)
|
|
|
|
actions = []
|
|
for call in job.get("calls") or []:
|
|
method = str(call.get("method") or "")
|
|
params = call.get("params") or {}
|
|
|
|
if method.lower() in ("switch.set", "relay.set"):
|
|
state = params.get("on")
|
|
if state is None:
|
|
state = params.get("turn")
|
|
channel = params.get("id", params.get("channel", 0))
|
|
|
|
if isinstance(state, str):
|
|
label = state.upper()
|
|
elif state is None:
|
|
label = "toggle"
|
|
else:
|
|
label = "ON" if state else "OFF"
|
|
|
|
actions.append(f"turn channel {channel} {label}")
|
|
elif method:
|
|
actions.append(method)
|
|
|
|
action_text = ", ".join(actions) if actions else "unknown action"
|
|
enabled = job.get("enable", job.get("enabled", True))
|
|
state = "" if enabled else " [disabled]"
|
|
return f"{when}: {action_text}{state}", bool(enabled)
|
|
|
|
|
|
async def rpc(session, host, method, params=None, auth=None):
|
|
url = f"http://{host}/rpc/{method}"
|
|
try:
|
|
async with session.post(
|
|
url,
|
|
json=params or {},
|
|
timeout=aiohttp.ClientTimeout(total=CONTROL_TIMEOUT),
|
|
auth=auth,
|
|
) as response:
|
|
if response.status != 200:
|
|
return None
|
|
return await response.json(content_type=None)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def fetch_automation(session, host, gen, config=None, auth=None):
|
|
found = {"schedules": [], "webhooks": [], "timers": {}, "checked": True}
|
|
|
|
if gen >= 2:
|
|
schedules = await rpc(session, host, "Schedule.List", auth=auth)
|
|
for job in (schedules or {}).get("jobs") or []:
|
|
text, enabled = describe_job(job)
|
|
found["schedules"].append(
|
|
{"id": job.get("id"), "description": text, "enabled": enabled}
|
|
)
|
|
|
|
hooks = await rpc(session, host, "Webhook.List", auth=auth)
|
|
for hook in (hooks or {}).get("hooks") or []:
|
|
found["webhooks"].append(
|
|
{
|
|
"id": hook.get("id"),
|
|
"event": hook.get("event", "?"),
|
|
"name": hook.get("name") or "",
|
|
"enabled": bool(hook.get("enable", True)),
|
|
}
|
|
)
|
|
|
|
for key, entry in (config or {}).items():
|
|
if not key.startswith("switch:"):
|
|
continue
|
|
index = key.split(":", 1)[1]
|
|
timers = {}
|
|
if entry.get("auto_on"):
|
|
timers["auto_on_after"] = entry.get("auto_on_delay")
|
|
if entry.get("auto_off"):
|
|
timers["auto_off_after"] = entry.get("auto_off_delay")
|
|
if entry.get("initial_state") not in (None, "restore_last"):
|
|
timers["initial_state"] = entry.get("initial_state")
|
|
if timers:
|
|
found["timers"][index] = timers
|
|
return found
|
|
|
|
relays = (config or {}).get("relays") or []
|
|
for index, relay in enumerate(relays):
|
|
if relay.get("schedule"):
|
|
for rule in relay.get("schedule_rules") or []:
|
|
found["schedules"].append(
|
|
{"description": f"relay {index}: {rule}", "enabled": True}
|
|
)
|
|
timers = {}
|
|
if relay.get("auto_on"):
|
|
timers["auto_on_after"] = relay.get("auto_on")
|
|
if relay.get("auto_off"):
|
|
timers["auto_off_after"] = relay.get("auto_off")
|
|
if timers:
|
|
found["timers"][str(index)] = timers
|
|
|
|
for action, hooks in ((config or {}).get("actions") or {}).get("active", {}).items():
|
|
found["webhooks"].append({"event": action, "name": "", "enabled": True})
|
|
|
|
return found
|
|
|
|
|
|
async def delete_schedule(session, host, job_id, gen, auth=None):
|
|
if gen >= 2:
|
|
result = await rpc(session, host, "Schedule.Delete", {"id": job_id}, auth)
|
|
return result is not None
|
|
return False
|
|
|
|
|
|
async def delete_webhook(session, host, hook_id, gen, auth=None):
|
|
if gen >= 2:
|
|
result = await rpc(session, host, "Webhook.Delete", {"id": hook_id}, auth)
|
|
return result is not None
|
|
return False
|
|
|
|
|
|
async def set_initial_state(session, host, channel, state, gen, auth=None):
|
|
if gen >= 2:
|
|
result = await rpc(
|
|
session,
|
|
host,
|
|
"Switch.SetConfig",
|
|
{"id": int(channel), "config": {"initial_state": state}},
|
|
auth,
|
|
)
|
|
return result is not None
|
|
return False
|
|
|
|
|
|
async def clear_auto_timer(session, host, channel, which, gen, auth=None):
|
|
if gen < 2:
|
|
return False
|
|
key = "auto_on" if which == "auto_on_after" else "auto_off"
|
|
result = await rpc(
|
|
session,
|
|
host,
|
|
"Switch.SetConfig",
|
|
{"id": int(channel), "config": {key: False}},
|
|
auth,
|
|
)
|
|
return result is not None
|
|
|
|
|
|
def automation_warnings(automation, channel=None):
|
|
warnings = []
|
|
if not automation:
|
|
return warnings
|
|
|
|
for job in automation.get("schedules") or []:
|
|
if job.get("enabled"):
|
|
warnings.append(f"schedule on the device: {job['description']}")
|
|
|
|
for index, timers in (automation.get("timers") or {}).items():
|
|
if channel is not None and str(channel) != str(index):
|
|
continue
|
|
for key, value in timers.items():
|
|
if key == "initial_state":
|
|
if str(value).lower() != "off":
|
|
continue
|
|
warnings.append(
|
|
f"channel {index} powers up to 'off' instead of restoring its "
|
|
"last state. After a power cut this plug stays OFF, so anything "
|
|
"it charges will not recover on its own"
|
|
)
|
|
else:
|
|
warnings.append(
|
|
f"channel {index} has {key} = {value}s, which will undo "
|
|
"commands on its own"
|
|
)
|
|
|
|
for hook in automation.get("webhooks") or []:
|
|
if hook.get("enabled"):
|
|
label = hook.get("name") or hook.get("event")
|
|
warnings.append(f"webhook/action on the device: {label}")
|
|
|
|
return warnings
|
|
|
|
|
|
def extract_channels(payload, status, config=None):
|
|
gen = payload.get("_gen", 1)
|
|
channels = {}
|
|
|
|
if gen >= 2:
|
|
for key, value in (status or {}).items():
|
|
if key.startswith("switch:"):
|
|
index = key.split(":", 1)[1]
|
|
channels[index] = {
|
|
"id": int(index),
|
|
"name": channel_name_from(config, gen, int(index))
|
|
or value.get("name")
|
|
or f"switch {index}",
|
|
"has_power_meter": "apower" in value,
|
|
}
|
|
if not channels:
|
|
channels["0"] = {"id": 0, "name": "switch 0", "has_power_meter": False}
|
|
return channels
|
|
|
|
relays = (status or {}).get("relays") or []
|
|
count = len(relays) or int(payload.get("num_outputs", 1) or 1)
|
|
meters = (status or {}).get("meters") or []
|
|
for index in range(count):
|
|
channels[str(index)] = {
|
|
"id": index,
|
|
"name": channel_name_from(config, gen, index) or f"relay {index}",
|
|
"has_power_meter": index < len(meters),
|
|
}
|
|
return channels
|
|
|
|
|
|
def flatten_status(status, prefix="", out=None, depth=0):
|
|
if out is None:
|
|
out = {}
|
|
if depth > 3:
|
|
return out
|
|
if isinstance(status, dict):
|
|
for key, value in status.items():
|
|
label = f"{prefix}{key}" if not prefix else f"{prefix}.{key}"
|
|
if isinstance(value, (dict, list)):
|
|
flatten_status(value, label, out, depth + 1)
|
|
else:
|
|
out[label] = value
|
|
elif isinstance(status, list):
|
|
for index, value in enumerate(status):
|
|
label = f"{prefix}[{index}]"
|
|
if isinstance(value, (dict, list)):
|
|
flatten_status(value, label, out, depth + 1)
|
|
else:
|
|
out[label] = value
|
|
return out
|
|
|
|
|
|
def find_duplicates(identifier, mac, host, keep):
|
|
import yaml
|
|
|
|
duplicates = []
|
|
if not paths.SHELLY_PROFILE_DIR.exists():
|
|
return duplicates
|
|
|
|
markers = {str(v).lower() for v in (identifier, mac, host) if v}
|
|
|
|
for candidate in sorted(paths.SHELLY_PROFILE_DIR.iterdir()):
|
|
if candidate.suffix not in (".yaml", ".yml") or candidate == keep:
|
|
continue
|
|
try:
|
|
with candidate.open("r", encoding="utf-8") as handle:
|
|
data = yaml.safe_load(handle)
|
|
except Exception:
|
|
continue
|
|
if not isinstance(data, dict):
|
|
continue
|
|
|
|
identity = data.get("identity") or {}
|
|
access = data.get("access") or {}
|
|
found = {
|
|
str(identity.get("id") or "").lower(),
|
|
str(identity.get("mac") or "").lower(),
|
|
str(access.get("host") or "").lower(),
|
|
}
|
|
if markers & (found - {""}):
|
|
duplicates.append(candidate)
|
|
|
|
return duplicates
|
|
|
|
|
|
def build_profile(payload, status, config=None, automation=None):
|
|
gen = payload.get("_gen", 1)
|
|
host = payload.get("_host")
|
|
|
|
if gen >= 2:
|
|
identifier = payload.get("id") or payload.get("mac")
|
|
model = payload.get("model") or payload.get("app") or "unknown"
|
|
else:
|
|
identifier = payload.get("mac")
|
|
model = payload.get("type") or "unknown"
|
|
|
|
name = device_name_from(payload, config, gen)
|
|
|
|
readable = {}
|
|
for key, value in sorted(flatten_status(status).items()):
|
|
readable[key] = value
|
|
|
|
aliases = []
|
|
for candidate in (name, identifier, payload.get("mac"), host):
|
|
if candidate and str(candidate) not in aliases:
|
|
aliases.append(str(candidate))
|
|
|
|
return {
|
|
"kind": "shelly",
|
|
"generated": now_iso(),
|
|
"aliases": aliases,
|
|
"identity": {
|
|
"id": identifier,
|
|
"mac": payload.get("mac"),
|
|
"make": "Shelly",
|
|
"model": model,
|
|
"name": name,
|
|
"generation": gen,
|
|
"firmware": payload.get("ver") or payload.get("fw") or "",
|
|
},
|
|
"access": {
|
|
"transport": "local-http",
|
|
"host": host,
|
|
"engine_mode": "read-write",
|
|
},
|
|
"auth": {
|
|
"required": bool(payload.get("auth") or payload.get("auth_en")),
|
|
"username": "",
|
|
"password": "",
|
|
},
|
|
"channels": extract_channels(payload, status, config),
|
|
"device_automation": automation or {},
|
|
"readable": readable,
|
|
}
|
|
|
|
|
|
async def discover(hosts=None, network=None, use_mdns=True, verbose=True):
|
|
paths.ensure_dirs()
|
|
|
|
candidates = []
|
|
|
|
if hosts:
|
|
candidates.extend(hosts)
|
|
else:
|
|
if use_mdns:
|
|
candidates.extend(mdns_discover(verbose=verbose))
|
|
|
|
found = []
|
|
if candidates:
|
|
found.extend(await probe_hosts(sorted(set(candidates)), verbose=verbose))
|
|
|
|
if not hosts:
|
|
if network:
|
|
networks = [network]
|
|
else:
|
|
networks = local_subnets()
|
|
if verbose and len(networks) > 1:
|
|
print(
|
|
f"Detected {len(networks)} local subnet(s): "
|
|
+ ", ".join(str(item) for item in networks)
|
|
)
|
|
if not networks and verbose:
|
|
print(
|
|
"Could not detect a local subnet. Use --network 192.168.1.0/24 "
|
|
"or --host <address>."
|
|
)
|
|
|
|
for target_network in networks:
|
|
already = {item["_host"] for item in found}
|
|
scanned = await scan_network(target_network, verbose=verbose)
|
|
found.extend(item for item in scanned if item["_host"] not in already)
|
|
|
|
written = []
|
|
used_names = {}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
for payload in found:
|
|
host = payload["_host"]
|
|
gen = payload["_gen"]
|
|
status = await fetch_status(session, host, gen)
|
|
config = await fetch_config(session, host, gen)
|
|
automation = await fetch_automation(session, host, gen, config)
|
|
profile = build_profile(payload, status, config, automation)
|
|
|
|
identity = profile["identity"]
|
|
identifier = identity["id"] or identity["mac"] or host
|
|
friendly = identity.get("name") or ""
|
|
|
|
if friendly:
|
|
stem = slugify(friendly).lower()
|
|
if stem in used_names:
|
|
suffix = slugify(str(identifier))[-6:]
|
|
stem = f"{stem}-{suffix}"
|
|
print(
|
|
f" note: two devices are named {friendly!r}, "
|
|
f"using {stem}.yaml for this one"
|
|
)
|
|
used_names[stem] = host
|
|
else:
|
|
stem = f"{slugify(identity['model'])}-{slugify(str(identifier))}".lower()
|
|
print(
|
|
f" note: {host} has no friendly name set. Name it in the Shelly "
|
|
"app and rerun discovery for a readable filename."
|
|
)
|
|
|
|
destination = paths.SHELLY_PROFILE_DIR / f"{stem}.yaml"
|
|
|
|
stale = find_duplicates(identifier, identity.get("mac"), host, destination)
|
|
|
|
save_yaml(destination, profile, header=PROFILE_HEADER)
|
|
written.append(destination)
|
|
|
|
for warning in automation_warnings(automation):
|
|
print(f" CONFLICT RISK: {warning}")
|
|
|
|
for other in stale:
|
|
print(
|
|
f" WARNING: {other.name} also describes this device. "
|
|
"Two profiles for one plug will confuse power profiles."
|
|
)
|
|
print(f" rm {other}")
|
|
|
|
if verbose:
|
|
label = f"{friendly} " if friendly else ""
|
|
print(
|
|
f" wrote {paths.relative(destination)} "
|
|
f"{label}({len(profile['channels'])} channel(s))"
|
|
)
|
|
|
|
return written
|
|
|
|
|
|
class ShellyTarget:
|
|
def __init__(self, profile_path, channel=None):
|
|
self.profile_path = Path(profile_path)
|
|
self.profile = load_yaml(self.profile_path)
|
|
identity = self.profile.get("identity", {})
|
|
access = self.profile.get("access", {})
|
|
|
|
self.host = access.get("host")
|
|
self.generation = int(identity.get("generation", 1) or 1)
|
|
self.model = identity.get("model", "unknown")
|
|
self.label = f"{self.model} @ {self.host}"
|
|
self.auth = auth_from(self.profile)
|
|
|
|
channels = self.profile.get("channels") or {}
|
|
if channel is None:
|
|
channel = sorted(channels)[0] if channels else "0"
|
|
self.channel = int(channel)
|
|
|
|
if not self.host:
|
|
raise ValueError(f"{self.profile_path} has no access.host")
|
|
|
|
async def set_state(self, session, on):
|
|
if self.generation >= 2:
|
|
url = f"http://{self.host}/rpc/Switch.Set"
|
|
payload = {"id": self.channel, "on": bool(on)}
|
|
async with session.post(
|
|
url,
|
|
json=payload,
|
|
timeout=aiohttp.ClientTimeout(total=CONTROL_TIMEOUT),
|
|
auth=self.auth,
|
|
) as response:
|
|
if response.status != 200:
|
|
raise RuntimeError(f"HTTP {response.status} from {url}")
|
|
return await response.json(content_type=None)
|
|
|
|
turn = "on" if on else "off"
|
|
url = f"http://{self.host}/relay/{self.channel}?turn={turn}"
|
|
async with session.get(
|
|
url, timeout=aiohttp.ClientTimeout(total=CONTROL_TIMEOUT), auth=self.auth
|
|
) as response:
|
|
if response.status != 200:
|
|
raise RuntimeError(f"HTTP {response.status} from {url}")
|
|
return await response.json(content_type=None)
|
|
|
|
async def get_state(self, session):
|
|
status = await fetch_status(session, self.host, self.generation, self.auth)
|
|
if not status:
|
|
return None
|
|
if self.generation >= 2:
|
|
entry = status.get(f"switch:{self.channel}") or {}
|
|
return entry.get("output")
|
|
relays = status.get("relays") or []
|
|
if self.channel < len(relays):
|
|
return relays[self.channel].get("ison")
|
|
return None
|
|
|
|
async def automation(self, session):
|
|
config = await fetch_config(session, self.host, self.generation, self.auth)
|
|
return await fetch_automation(
|
|
session, self.host, self.generation, config, self.auth
|
|
)
|
|
|
|
async def conflicts(self, session):
|
|
return automation_warnings(await self.automation(session), self.channel)
|
|
|
|
async def reachable(self, session):
|
|
try:
|
|
return await self.get_state(session) is not None
|
|
except Exception:
|
|
return False
|