shd/local_yandex_station
1
0
Fork 0
Code Issues Releases Wiki Activity

cleanup cli, configs, systemd and readme

Browse Source
This commit is contained in:
Mark Kats
2026-04-02 01:18:15 +03:00
parent 7a86519314
commit 50961eb3fc
17 changed files with 741 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
alice_plugin.py
View File

@@ -24,7 +24,9 @@ import yastation
BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
LOXONE_DIR = BASE_DIR / "loxone"
CONFIGS_DIR = BASE_DIR / "configs"
LOXONE_DIR = CONFIGS_DIR / "loxone"
WB_RULES_DIR = CONFIGS_DIR / "wb-rules"
STATIONS_PATH = DATA_DIR / "stations.json"
CONFIG_PATH = DATA_DIR / "config.json"
TOKEN_PATH = BASE_DIR / "token.txt"
@@ -59,7 +61,9 @@ class StationRecord:
def ensure_dirs() -> None:
DATA_DIR.mkdir(parents=True, exist_ok=True)
CONFIGS_DIR.mkdir(parents=True, exist_ok=True)
LOXONE_DIR.mkdir(parents=True, exist_ok=True)
WB_RULES_DIR.mkdir(parents=True, exist_ok=True)
def safe_name(value: str, fallback: str) -> str:
@@ -533,19 +537,17 @@ defineRule(DEVICE_ID + "_status_poll", {{
"""
def write_station_templates(station: StationRecord, controller_host: str, target_dir: Path) -> Dict[str, str]:
station_folder = target_dir / safe_name(station.name or station.station_id, station.station_id or "station")
def write_loxone_templates(station: StationRecord, controller_host: str) -> Dict[str, str]:
station_folder = LOXONE_DIR / safe_name(station.name or station.station_id, station.station_id or "station")
station_folder.mkdir(parents=True, exist_ok=True)
selector = station.selector()
vo_path = station_folder / "VO.xml"
vi_path = station_folder / "VI.xml"
wb_path = station_folder / "wb-rules-station.js"
readme_path = station_folder / "README.md"
vo_path.write_text(build_vo_xml(controller_host, station.name or station.station_id, selector), encoding="utf-8")
vi_path.write_text(build_vi_xml(controller_host, station.name or station.station_id, selector), encoding="utf-8")
wb_path.write_text(build_wb_rules_js(controller_host, station.name or station.station_id, selector), encoding="utf-8")
readme_path.write_text(
"# Alice templates\n\n"
f"- Station: {station.name}\n"
@@ -561,6 +563,31 @@ def write_station_templates(station: StationRecord, controller_host: str, target
"folder": str(station_folder.relative_to(BASE_DIR)),
"vo": str(vo_path.relative_to(BASE_DIR)),
"vi": str(vi_path.relative_to(BASE_DIR)),
}
def write_wb_rules_templates(station: StationRecord, controller_host: str) -> Dict[str, str]:
station_folder = WB_RULES_DIR / safe_name(station.name or station.station_id, station.station_id or "station")
station_folder.mkdir(parents=True, exist_ok=True)
selector = station.selector()
wb_path = station_folder / "wb-rules-station.js"
readme_path = station_folder / "README.md"
wb_path.write_text(build_wb_rules_js(controller_host, station.name or station.station_id, selector), encoding="utf-8")
readme_path.write_text(
"# Alice wb-rules templates\n\n"
f"- Station: {station.name}\n"
f"- Station ID: {station.station_id}\n"
f"- Selector: {selector}\n"
f"- API: http://{controller_host}:9124\n",
encoding="utf-8",
)
return {
"station": station.station_id,
"name": station.name,
"folder": str(station_folder.relative_to(BASE_DIR)),
"wb": str(wb_path.relative_to(BASE_DIR)),
}
@@ -573,28 +600,33 @@ def generate_templates(template_kind: str, controller_host: str) -> Dict[str, An
ensure_dirs()
result: List[Dict[str, str]] = []
for station in stations:
files = write_station_templates(station, controller_host, LOXONE_DIR)
if template_kind == "loxone":
files = {k: v for k, v in files.items() if k in {"station", "name", "folder", "vo", "vi"}}
elif template_kind == "wb-rules":
files = {k: v for k, v in files.items() if k in {"station", "name", "folder", "wb"}}
result.append(files)
if template_kind == "loxone":
target_dir = LOXONE_DIR
zip_name = "loxone_templates.zip"
for station in stations:
result.append(write_loxone_templates(station, controller_host))
elif template_kind == "wb-rules":
target_dir = WB_RULES_DIR
zip_name = "wb_rules_templates.zip"
for station in stations:
result.append(write_wb_rules_templates(station, controller_host))
else:
raise RuntimeError(f"unsupported template kind: {template_kind}")
zip_name = "loxone_templates.zip" if template_kind == "loxone" else "wb_rules_templates.zip"
zip_path = LOXONE_DIR / zip_name
zip_path = target_dir / zip_name
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
for station in stations:
folder = LOXONE_DIR / safe_name(station.name or station.station_id, station.station_id or "station")
if template_kind in ("loxone", "all"):
folder = target_dir / safe_name(station.name or station.station_id, station.station_id or "station")
if template_kind == "loxone":
for fn in ["VO.xml", "VI.xml", "README.md"]:
p = folder / fn
if p.exists():
zf.write(p, arcname=f"{folder.name}/{fn}")
if template_kind in ("wb-rules", "all"):
p = folder / "wb-rules-station.js"
if p.exists():
zf.write(p, arcname=f"{folder.name}/wb-rules-station.js")
elif template_kind == "wb-rules":
for fn in ["wb-rules-station.js", "README.md"]:
p = folder / fn
if p.exists():
zf.write(p, arcname=f"{folder.name}/{fn}")
return {"ok": True, "templates": result, "zip": str(zip_path.relative_to(BASE_DIR))}
@@ -616,10 +648,10 @@ def install_wb_rules(station_id: Optional[str] = None) -> Dict[str, Any]:
installed: List[str] = []
for station in stations:
folder = LOXONE_DIR / safe_name(station.name or station.station_id, station.station_id or "station")
folder = WB_RULES_DIR / safe_name(station.name or station.station_id, station.station_id or "station")
src = folder / "wb-rules-station.js"
if not src.exists():
write_station_templates(station, detect_controller_host(), LOXONE_DIR)
write_wb_rules_templates(station, detect_controller_host())
src = folder / "wb-rules-station.js"
dst = target_dir / f"alice_station_{safe_name(station.station_id, 'station')}.js"
@@ -643,7 +675,14 @@ async def refresh_stations(token: str, timeout: float = 3.0) -> Dict[str, Any]:
if not token:
raise RuntimeError("token is required")
cloud = await fetch_cloud_stations(token)
try:
cloud = await fetch_cloud_stations(token)
except Exception as exc:
msg = str(exc)
if "AUTH_TOKEN_INVALID" in msg:
raise RuntimeError("YANDEX_OAUTH_TOKEN invalid: update token and retry") from exc
raise
local = await discover_local(timeout=timeout)
merged = merge_cloud_with_local(cloud, local)
@@ -728,7 +767,7 @@ async def api_download(request: web.Request) -> web.StreamResponse:
if kind not in {"loxone", "wb-rules"}:
raise web.HTTPNotFound()
zip_name = "loxone_templates.zip" if kind == "loxone" else "wb_rules_templates.zip"
zip_path = LOXONE_DIR / zip_name
zip_path = (LOXONE_DIR if kind == "loxone" else WB_RULES_DIR) / zip_name
if not zip_path.exists():
generate_templates(kind, detect_controller_host())
return web.FileResponse(zip_path)
@@ -779,21 +818,21 @@ def build_parser() -> argparse.ArgumentParser:
sub = p.add_subparsers(dest="cmd", required=True)
sp = sub.add_parser("stations-refresh", help="1) get stations list")
sp.add_argument("--timeout", type=float, default=3.0)
sp = sub.add_parser("stations-refresh", help="Обновить список станций")
sp.add_argument("--timeout", type=float, default=3.0, help="Секунды ожидания локального mDNS-поиска")
sp = sub.add_parser("templates-loxone", help="2) generate Loxone templates in ./loxone")
sp.add_argument("--controller-host", default="")
sp = sub.add_parser("templates-loxone", help="Сгенерировать шаблоны Loxone в ./configs/loxone")
sp.add_argument("--controller-host", default="", help="IP или host контроллера для ссылок на API")
sp = sub.add_parser("templates-wb-rules", help="3) generate wb-rules templates in ./loxone")
sp.add_argument("--controller-host", default="")
sp = sub.add_parser("templates-wb-rules", help="Сгенерировать шаблоны wb-rules в ./configs/wb-rules")
sp.add_argument("--controller-host", default="", help="IP или host контроллера для ссылок на API")
sp = sub.add_parser("wb-install", help="install wb-rules templates to /etc/wb-rules")
sp.add_argument("--station-id", default="")
sp = sub.add_parser("wb-install", help="Установить wb-rules шаблоны в /etc/wb-rules")
sp.add_argument("--station-id", default="", help="Установить только одну станцию по station id")
sp = sub.add_parser("web", help="run web UI")
sp.add_argument("--host", default="0.0.0.0")
sp.add_argument("--port", type=int, default=9140)
sp = sub.add_parser("web", help="Запустить web UI")
sp.add_argument("--host", default="0.0.0.0", help="Хост для web UI")
sp.add_argument("--port", type=int, default=9140, help="Порт для web UI")
return p