"""Shared helpers for MowDeck spike 0001 (PyMammotion 0.10.9)."""

from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
import sys
import time
from pathlib import Path
from typing import Any

from pymammotion.client import MammotionClient
from pymammotion.data.model.device import MowingDevice
from pymammotion.transport.base import TransportType

OUT = Path(os.environ.get("SPIKE_OUT", "out"))
OUT.mkdir(exist_ok=True)
TOKEN_CACHE = OUT / "token_cache.json"
log = logging.getLogger("spike")


def setup_logging() -> None:
    root = logging.getLogger()
    root.setLevel(logging.DEBUG)
    fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
    console = logging.StreamHandler(sys.stdout)
    console.setLevel(logging.INFO)
    logfile = logging.FileHandler(OUT / "spike.log", encoding="utf-8")
    logfile.setLevel(logging.DEBUG)
    for handler in (console, logfile):
        handler.setFormatter(fmt)
        root.addHandler(handler)


def jsonable(value: Any) -> Any:
    if hasattr(value, "to_dict"):
        return jsonable(value.to_dict())
    if isinstance(value, dict):
        return {str(k): jsonable(v) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [jsonable(v) for v in value]
    return value


def write_json(name: str, data: Any) -> None:
    path = OUT / name
    path.write_text(json.dumps(jsonable(data), indent=2, default=str), encoding="utf-8")
    log.info("wrote %s", path)


def transport_args(parser: argparse.ArgumentParser) -> argparse.ArgumentParser:
    parser.add_argument("--via", choices=["cloud", "ble", "proxy"], default="cloud")
    parser.add_argument("--name", help="device name, e.g. Luba-VPxxxxxx (required for ble/proxy)")
    parser.add_argument("--address", help="mower BLE MAC (required for ble/proxy)")
    parser.add_argument("--proxy", default="mowdeck-bridge.local", help="ESPHome proxy host")
    return parser


async def connect_cloud() -> MammotionClient:
    email, password = os.environ["MAMMOTION_EMAIL"], os.environ["MAMMOTION_PASSWORD"]
    client = MammotionClient("mowdeck-spike-0001")

    async def save_cache() -> None:
        if cache := client.to_cache():
            TOKEN_CACHE.write_text(json.dumps(jsonable(cache)), encoding="utf-8")

    client.on_credentials_updated = save_cache
    restored = False
    if TOKEN_CACHE.exists():
        try:
            cached = json.loads(TOKEN_CACHE.read_text(encoding="utf-8"))
            await client.restore_credentials(email, password, cached)
            restored = True
            log.info("session restored from token cache")
        except Exception:
            log.warning("token cache restore failed, doing a password login", exc_info=True)
    if not restored:
        await client.login_and_initiate_cloud(email, password)
    await save_cache()
    await asyncio.sleep(3)  # let MQTT subscribe and first reports arrive
    return client


async def _register_ble(client: MammotionClient, name: str, **ble: Any) -> None:
    handle = await client.add_ble_only_device(
        device_id=name, device_name=name, initial_device=MowingDevice(name=name), **ble
    )
    await handle.start()
    transport = handle.get_transport(TransportType.BLE)
    if transport is None:
        raise RuntimeError("handle has no BLE transport")
    t0 = time.perf_counter()
    await transport.connect()
    log.info("BLE connected in %.2f s", time.perf_counter() - t0)


async def connect_ble_proxy(client: MammotionClient, name: str, address: str, host: str) -> None:
    from aioesphomeapi import APIClient
    from bleak_esphome import connect_scanner
    from bluetooth_adapters import get_adapters
    from habluetooth import BluetoothManager, set_manager

    class _NoOpManager(BluetoothManager):
        def _discover_service_info(self, service_info: object) -> None:
            return

    manager = _NoOpManager(get_adapters(), slot_manager=None)
    set_manager(manager)
    await manager.async_setup()
    api = APIClient(
        address=host, port=6053, password=None, noise_psk=os.environ["ESPHOME_NOISE_PSK"]
    )
    await api.connect(login=True)
    info = await api.device_info()
    log.info(
        "proxy %s connected, bt features %s",
        info.name,
        info.bluetooth_proxy_feature_flags_compat(api.api_version),
    )
    scanner = connect_scanner(api, info, available=True).scanner
    await scanner.async_setup()
    manager.async_register_scanner(scanner)
    loop = asyncio.get_running_loop()
    deadline, ble_device = loop.time() + 30, None
    while ble_device is None and loop.time() < deadline:
        for dev, _adv in scanner.discovered_devices_and_advertisement_data.values():
            if dev.address.upper() == address.upper():
                ble_device = dev
        await asyncio.sleep(0.5)
    if ble_device is None:
        raise RuntimeError(f"{address} not seen by proxy {host} within 30 s")
    await _register_ble(client, name, ble_device=ble_device, self_managed_scanning=False)


async def open_session(args: argparse.Namespace) -> tuple[MammotionClient, str]:
    if args.via == "cloud":
        client = await connect_cloud()
        names = [h.device_name for h in client.device_registry.all_devices]
        name = args.name or next((n for n in names if n.startswith("Luba")), None)
        if name is None:
            raise SystemExit(f"no Luba on this account, devices: {names}")
        return client, name
    if not (args.name and args.address):
        raise SystemExit("--name and --address are required for --via ble/proxy (see step 0)")
    client = MammotionClient("mowdeck-spike-0001")
    if args.via == "ble":
        await _register_ble(client, args.name, ble_address=args.address.upper())
    else:
        await connect_ble_proxy(client, args.name, args.address, args.proxy)
    await asyncio.sleep(2)
    return client, args.name


def transport_label(client: MammotionClient, name: str) -> str:
    handle = client.mower(name)
    if handle is None:
        return "none"
    for kind in (TransportType.BLE, TransportType.CLOUD_MAMMOTION, TransportType.CLOUD_ALIYUN):
        if handle.is_transport_connected(kind):
            return kind.value
    return "none"


async def wait_saga_idle(client: MammotionClient, name: str, timeout: float = 300.0) -> None:
    handle = client.mower(name)
    loop = asyncio.get_running_loop()
    start_by, deadline = loop.time() + 3, loop.time() + timeout
    while loop.time() < start_by and not handle.queue.is_saga_active:
        await asyncio.sleep(0.05)
    while handle.queue.is_saga_active:
        if loop.time() > deadline:
            raise TimeoutError("saga did not finish")
        await asyncio.sleep(0.5)

import asyncio
from bleak import BleakScanner


async def main() -> None:
    found = await BleakScanner.discover(timeout=15.0, return_adv=True)
    for address, (dev, adv) in found.items():
        name = adv.local_name or dev.name or ""
        if name.startswith(("Luba", "Yuka")):
            print(f"{address}  {name}  rssi={adv.rssi}  services={adv.service_uuids}")


asyncio.run(main())

import argparse, asyncio
from common import connect_cloud, setup_logging, transport_label, write_json


async def main() -> None:
    setup_logging()
    client = await connect_cloud()
    rows = []
    for h in client.device_registry.all_devices:
        rows.append(
            {
                "name": h.device_name,
                "iot_id": h.iot_id,
                "account": h.account_id,
                "firmware": h.firmware_version,
                "transport": transport_label(client, h.device_name),
            }
        )
        print(rows[-1])
    write_json("s01_devices.json", rows)
    await client.stop()


asyncio.run(main())

import argparse, asyncio
from pymammotion.data.model.enums import RTKStatus
from pymammotion.utility.constant.device_enums import WorkMode
from common import open_session, setup_logging, transport_args, transport_label, write_json


async def main() -> None:
    args = transport_args(argparse.ArgumentParser()).parse_args()
    setup_logging()
    client, name = await open_session(args)
    samples = []
    for _ in range(30):  # 60 s
        dev = client.get_device_by_name(name)
        rd, rtk = dev.report_data, dev.report_data.rtk
        s = {
            "transport": transport_label(client, name),
            "battery": rd.dev.battery_val,
            "charge_state": rd.dev.charge_state,
            "mode": WorkMode(rd.dev.sys_status).name,
            "rtk_status": RTKStatus.from_value(rtk.status).name,
            "pos_level": rtk.positioning_mode.name,
            "sats": rtk.gps_stars,
            "l2_sats": rtk.l2_stars,
            "corr_age_s": rtk.age,
            "correction_source": rtk.mqtt_rtk_info.switch_mode.name,
            "pos_deg": (dev.location.device.latitude, dev.location.device.longitude),
        }
        samples.append(s)
        print(s)
        await asyncio.sleep(2)
    write_json(f"s02_status_{args.via}.json", samples)
    await client.stop()


asyncio.run(main())

import argparse, asyncio
from pymammotion.data.model.generate_geojson import apply_area_geojson
from common import open_session, setup_logging, transport_args, wait_saga_idle, write_json


async def main() -> None:
    args = transport_args(argparse.ArgumentParser()).parse_args()
    setup_logging()
    client, name = await open_session(args)
    await client.start_map_sync(name)
    await wait_saga_idle(client, name, timeout=300)
    dev = client.get_device_by_name(name)
    m = dev.map
    write_json(
        "s03_map_raw.json",
        {
            "area": m.area,
            "obstacle": m.obstacle,
            "path": m.path,
            "no_go_zone": m.no_go_zone,
            "virtual_wall": m.virtual_wall,
            "dump": m.dump,
            "svg": m.svg,
            "area_name": m.area_name,
        },
    )
    write_json(
        "s03_frames.json",
        {
            "rtk_origin_rad": dev.location.RTK,
            "dock": dev.location.dock,
            "device_deg": dev.location.device,
            "rtk_info": dev.report_data.rtk.mqtt_rtk_info,
        },
    )
    if not m.generated_geojson:
        apply_area_geojson(m, dev.location.RTK, dev.location.dock)
    write_json("s03_map.geojson", m.generated_geojson)
    print({k: len(getattr(m, k)) for k in ("area", "obstacle", "path", "no_go_zone", "svg")})
    await client.stop()


asyncio.run(main())

import argparse, asyncio, time
from pymammotion.data.model.device_config import OperationSettings, build_route_information
from pymammotion.messaging.command_queue import Priority
from pymammotion.utility.constant.device_enums import WorkMode
from common import open_session, setup_logging, transport_args, transport_label, write_json


async def main() -> None:
    p = transport_args(argparse.ArgumentParser())
    p.add_argument("--zone", type=int, required=True)
    p.add_argument("--blade-height", type=int, default=60)
    args = p.parse_args()
    setup_logging()
    client, name = await open_session(args)
    log = []

    async def step(label, coro, mode=None):
        t0 = time.perf_counter()
        await coro
        sent = time.perf_counter() - t0
        if mode is not None:
            await client.wait_for(name, lambda d: d.report_data.dev.sys_status == mode, timeout=90)
        log.append(
            {
                "step": label,
                "send_s": round(sent, 3),
                "until_state_s": round(time.perf_counter() - t0, 3),
                "transport": transport_label(client, name),
            }
        )
        print(log[-1])

    dev = client.get_device_by_name(name)
    settings = OperationSettings(
        areas=[args.zone], blade_height=args.blade_height, channel_mode=0, toward=0
    )
    route = build_route_information(name, dev, settings)
    input("Area clear, someone at STOP? Press Enter to plan and start ")
    await step(
        "plan",
        client.send_command_and_wait(
            name,
            "generate_route_information",
            "bidire_reqconver_path",
            generate_route_information=route,
            priority=Priority.USER,
        ),
    )
    await step(
        "start",
        client.send_command_and_wait(
            name, "start_job", "zone_start_precent_t", priority=Priority.USER
        ),
        WorkMode.MODE_WORKING,
    )
    await asyncio.sleep(30)
    await step(
        "pause",
        client.send_command_with_args(name, "pause_execute_task", priority=Priority.USER),
        WorkMode.MODE_PAUSE,
    )
    await asyncio.sleep(10)
    await step(
        "resume",
        client.send_command_with_args(name, "resume_execute_task", priority=Priority.USER),
        WorkMode.MODE_WORKING,
    )
    await asyncio.sleep(10)
    await step(
        "dock",
        client.send_command_with_args(name, "return_to_dock", priority=Priority.USER),
        WorkMode.MODE_RETURNING,
    )
    write_json(f"s04_commands_{args.via}.json", log)
    await client.stop()


asyncio.run(main())

import argparse, asyncio, csv, os, statistics, time
import psutil
from pymammotion.messaging.command_queue import Priority
from common import OUT, open_session, setup_logging, transport_args, transport_label


async def main() -> None:
    p = transport_args(argparse.ArgumentParser())
    p.add_argument("-n", type=int, default=100)
    args = p.parse_args()
    setup_logging()
    client, name = await open_session(args)
    rows = []
    for i in range(args.n):
        for kind in ("rtt", "move0"):
            t0 = time.perf_counter()
            ok = True
            try:
                if kind == "rtt":
                    await client.send_command_and_wait(
                        name,
                        "get_device_version_info",
                        "toapp_dev_fw_info",
                        send_timeout=3.0,
                        priority=Priority.USER,
                    )
                else:
                    await client.send_command_with_args(
                        name,
                        "send_movement",
                        linear_speed=0,
                        angular_speed=0,
                        priority=Priority.USER,
                    )
            except Exception as exc:
                ok = False
                print(i, kind, "error", exc)
            rows.append(
                {
                    "i": i,
                    "kind": kind,
                    "ms": round((time.perf_counter() - t0) * 1000, 1),
                    "ok": ok,
                    "transport": transport_label(client, name),
                }
            )
        await asyncio.sleep(0.2)
    with open(OUT / f"s05_latency_{args.via}.csv", "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=rows[0].keys())
        w.writeheader()
        w.writerows(rows)
    for kind in ("rtt", "move0"):
        ms = sorted(r["ms"] for r in rows if r["kind"] == kind and r["ok"])
        if ms:
            print(
                kind,
                "n",
                len(ms),
                "p50",
                statistics.median(ms),
                "p95",
                ms[int(len(ms) * 0.95) - 1],
                "max",
                ms[-1],
            )
    proc = psutil.Process(os.getpid())
    print("rss_mb", round(proc.memory_info().rss / 1e6, 1), "cpu_s", sum(proc.cpu_times()[:2]))
    await client.stop()


asyncio.run(main())

import argparse, asyncio, csv, time
from pymammotion.messaging.command_queue import Priority
from common import OUT, open_session, setup_logging, transport_args

FRAME_S, DEADMAN_S = 0.15, 1.0
STOP = ("send_movement", {"linear_speed": 0, "angular_speed": 0})


async def main() -> None:
    args = transport_args(argparse.ArgumentParser()).parse_args()
    setup_logging()
    client, name = await open_session(args)
    rows, t_start = [], time.perf_counter()
    stick = {"cmd": STOP, "t": 0.0}  # last stick input and its time
    sender_on = True  # False simulates a dead bridge

    async def send(cmd, kw, label):
        t0 = time.perf_counter()
        await client.send_command_with_args(name, cmd, priority=Priority.USER, **kw)
        loc = client.get_device_by_name(name).location.device
        rows.append(
            {
                "t": round(t0 - t_start, 3),
                "label": label,
                "cmd": cmd,
                "ms": round((time.perf_counter() - t0) * 1000, 1),
                "lat": loc.latitude,
                "lon": loc.longitude,
            }
        )

    async def sender():  # the deadman loop, as the gateway would run it
        stopped = True
        while True:
            if sender_on:
                if time.perf_counter() - stick["t"] <= DEADMAN_S and stick["cmd"] is not STOP:
                    await send(*stick["cmd"], "drive")
                    stopped = False
                elif not stopped:
                    await send(*STOP, "deadman_stop")
                    stopped = True
            await asyncio.sleep(FRAME_S)

    async def hold(cmd, kw, seconds):  # simulated stick held, input every 100 ms
        end = time.perf_counter() + seconds
        while time.perf_counter() < end:
            stick.update(cmd=(cmd, kw), t=time.perf_counter())
            await asyncio.sleep(0.1)

    task = asyncio.create_task(sender())
    input("Lawn clear, blades off, hand on STOP? Enter for A: forward 2 s, then release ")
    await hold("move_forward", {"linear": 0.35}, 2.0)
    stick["cmd"] = STOP
    await send(*STOP, "release")
    input(
        "Enter for B: left turn 2 s, then input stops without release (deadman must stop it after 1 s) "
    )
    await hold("move_left", {"angular": 0.35}, 2.0)
    await asyncio.sleep(2.0)
    input(
        "Enter for C (dead bridge): forward 2 s, then NO frames for 5 s. Watch if it keeps rolling "
    )
    await hold("move_forward", {"linear": 0.35}, 2.0)
    sender_on = False
    await asyncio.sleep(5.0)
    stick["cmd"] = STOP
    sender_on = True
    await send(*STOP, "safety_stop")
    task.cancel()
    with open(OUT / f"s06_drive_{args.via}.csv", "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=rows[0].keys())
        w.writeheader()
        w.writerows(rows)
    await client.stop()


asyncio.run(main())

import asyncio
from common import connect_cloud, setup_logging, write_json


async def main() -> None:
    setup_logging()
    client = await connect_cloud()
    handle = next(h for h in client.device_registry.all_devices if h.device_name.startswith("Luba"))
    resp = await client.get_stream_subscription(handle.device_name, handle.iot_id)
    data = resp.data if resp is not None else None
    video = await client.mammotion_http.get_video_resource(handle.iot_id)
    summary = {
        "code": getattr(resp, "code", None),
        "has_token": bool(data and data.token),
        "channel": bool(data and data.channelName),
        "open_encrypt": data.openEncrypt if data else None,
        "cameras": len(data.cameras) if data else 0,
        "available_time": data.availableTime if data else None,
        "video_quota": video.data if video else None,
    }
    print(summary)
    write_json("s07_camera.json", summary)
    await asyncio.sleep(5)
    await client.stop_stream(handle.device_name)
    await client.stop()


asyncio.run(main())

import argparse, asyncio
from common import open_session, setup_logging, transport_args, write_json


async def main() -> None:
    args = transport_args(argparse.ArgumentParser()).parse_args()
    setup_logging()
    client, name = await open_session(args)
    await client.refresh_status(name)
    await asyncio.sleep(10)
    dev, handle = client.get_device_by_name(name), client.mower(name)
    rd, rtk = dev.report_data, dev.report_data.rtk
    mnet = rd.dev.mnet_info
    out = {
        "connect": rd.connect,
        "sim": mnet.sim,
        "link_type": mnet.link_type,
        "mnet_rssi": mnet.rssi,
        "operator": mnet.operator,
        "inet_connected": mnet.inet.connect,
        "fpv_info": rd.dev.fpv_info,
        "rtk_switch": rtk.mqtt_rtk_info.rtk_switch,
        "nrtk_net_mode": rtk.mqtt_rtk_info.nrtk_net_mode,
        "nrtk_map_convert_status": rtk.mqtt_rtk_info.nrtk_map_convert_status,
        "dis_status": rtk.get_dis_status(),
        "score": rtk.score_info,
        "supports_wifi_movement": handle.supports_wifi_movement(),
    }
    print(out)
    write_json(f"s08_net_{args.via}.json", out)  # IMEI/ICCID are not written
    await client.stop()


asyncio.run(main())