fgl-aircon/pyfglair/__main__.py

"""CLI pyfglair: discover (облако) и monitor (LAN-сессия).

    python -m pyfglair discover --region eu --email user@example.com
    python -m pyfglair discover ... --format esphome-secrets
    python -m pyfglair monitor --config config_home.json --duration 60
"""
from __future__ import annotations

import argparse
import asyncio
import getpass
import json
import os
import sys
from typing import Any, Optional

from .const import State, Template, ValueKind
from .provision import REGIONS


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="pyfglair",
        description="FGLair (Fujitsu General) LAN-управление: provisioning и "
        "монитор сессии",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    disc = sub.add_parser("discover", help="облачный discovery (lanip_key)")
    disc.add_argument("--region", choices=sorted(REGIONS), default="eu")
    disc.add_argument("--email", default=os.environ.get("FGLAIR_EMAIL"))
    disc.add_argument("--password", default=os.environ.get("FGLAIR_PASSWORD"))
    disc.add_argument("--device", help="фильтр по имени устройства")
    disc.add_argument("--format", choices=["json", "esphome-secrets"],
                      default="json")
    disc.add_argument("--out", help="записать в файл (права 0600)")
    disc.add_argument("--insecure", action="store_true",
                      help="не проверять TLS-сертификаты")
    disc.add_argument("--api-base",
                      help="ВНУТРЕННЕЕ: базовый URL мок-облака (тесты)")

    mon = sub.add_parser("monitor", help="монитор LAN-сессии (события)")
    mon.add_argument("--config", help="config_*.json (fglair-discover/fglctl)")
    mon.add_argument("--host")
    mon.add_argument("--dsn")
    mon.add_argument("--lanip-key")
    mon.add_argument("--lanip-key-id", type=int)
    mon.add_argument("--template", choices=["A", "B", "F"],
                     help="шаблон (по умолчанию — по model, иначе A)")
    mon.add_argument("--device-port", type=int, default=80)
    mon.add_argument("--listen-port", type=int, default=10275)
    mon.add_argument("--keepalive-ms", type=int, default=15000)
    mon.add_argument("--duration", type=float, default=60.0)
    mon.add_argument("--get", help="свойства через запятую (enum или имя)")
    mon.add_argument("--log-level", type=int, default=1)
    return parser


def _load_monitor_config(args: argparse.Namespace) -> dict[str, Any]:
    if args.config:
        with open(args.config, "r", encoding="utf-8") as fh:
            data = json.load(fh)
        data.setdefault("ip_address", data.get("host"))
        return data
    if not (args.host and args.dsn and args.lanip_key
            and args.lanip_key_id is not None):
        raise SystemExit(
            "monitor: нужен --config или --host/--dsn/--lanip-key/"
            "--lanip-key-id"
        )
    return {
        "ip_address": args.host,
        "dsn": args.dsn,
        "lanip_key": args.lanip_key,
        "lanip_key_id": args.lanip_key_id,
        "model": args.template,
    }


def _resolve_template(data: dict[str, Any], explicit: Optional[str]) -> Template:
    name = explicit or None
    model = data.get("model")
    if name is None and model in ("A", "B", "F"):
        name = model
    if name is not None:
        return Template[name]
    if model:
        from .templates import detect

        found = detect(model)
        if found is None:
            print(f"модель {model} неизвестна ядру — шаблон A",
                  file=sys.stderr)
            return Template.A
        return found
    return Template.A


async def _run_discover(args: argparse.Namespace) -> int:
    from .provision import discover, esphome_secrets

    email = args.email
    if not email:
        raise SystemExit("discover: укажите --email или FGLAIR_EMAIL")
    password = args.password or getpass.getpass("Пароль FGLair: ")

    devices = await discover(
        email, password, args.region, base_url=args.api_base,
        insecure=args.insecure,
    )
    if args.device:
        devices = [d for d in devices if d.name == args.device]
        if not devices:
            raise SystemExit(f"discover: устройство {args.device!r} не найдено")

    if args.format == "esphome-secrets":
        text = esphome_secrets(devices)
    else:
        text = "\n".join(
            json.dumps(d.as_config(), ensure_ascii=False) for d in devices
        ) + "\n"

    if args.out:
        fd = os.open(args.out, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            fh.write(text)
        print(f"Записано: {args.out} (права 0600)", file=sys.stderr)
    else:
        sys.stdout.write(text)
    return 0


def _fmt_value(kind: ValueKind, value: int, text: str) -> str:
    if kind == ValueKind.STRING:
        return f'"{text}"'
    suffix = " (bool)" if kind == ValueKind.BOOL else ""
    return f"{value}{suffix}"


async def _run_monitor(args: argparse.Namespace) -> int:
    from .session import Config, Session, set_log_level
    from .templates import template_info

    data = _load_monitor_config(args)
    host = data.get("ip_address")
    dsn = data.get("dsn")
    key = data.get("lanip_key")
    key_id = data.get("lanip_key_id")
    if not (host and dsn and key and key_id is not None):
        raise SystemExit("monitor: в config нет ip_address/dsn/lanip_key/"
                         "lanip_key_id")

    template = _resolve_template(data, args.template)
    cfg = Config(
        host=host,
        dsn=dsn,
        lanip_key=key,
        lanip_key_id=int(key_id),
        template=template,
        device_port=args.device_port,
        listen_port=args.listen_port,
        keepalive_ms=args.keepalive_ms,
    )

    def on_state(state: State, error) -> None:
        print(f"STATE {state.name.lower()} err={int(error)}", flush=True)

    def on_property(ev) -> None:
        print(
            f"PROP {ev.prop.name} = "
            f"{_fmt_value(ev.value.kind, ev.value.int_value, ev.value.str_value)}"
            + (f"  [cmd_id={ev.cmd_id} status={ev.status}]"
               if ev.cmd_id >= 0 else ""),
            flush=True,
        )

    set_log_level(args.log_level)
    session = Session(cfg, on_state=on_state, on_property=on_property)
    if not session.start():
        print("monitor: не удалось запустить сессию", file=sys.stderr)
        session.close()
        return 2

    if args.get:
        def normalize(name: str) -> str:
            return name.strip().lower().replace("_", "")

        wanted = {normalize(name) for name in args.get.split(",")}
        for info in template_info(template):
            if normalize(info.name) in wanted or normalize(info.prop.name) in wanted:
                session.get_prop(info.prop)

    try:
        await asyncio.sleep(args.duration)
    finally:
        await session.async_stop()
    return 0


def main(argv: Optional[list[str]] = None) -> int:
    args = _build_parser().parse_args(argv)
    if args.command == "discover":
        return asyncio.run(_run_discover(args))
    if args.command == "monitor":
        try:
            return asyncio.run(_run_monitor(args))
        except KeyboardInterrupt:
            return 130
    return 2


if __name__ == "__main__":
    sys.exit(main())