#!/usr/bin/env python3
# Copyright (C) 2026 Christian Wirtz doc@snowheaven.de
# License: GPL-2.0-only, see LICENSE in the repository root.
"""Special agent for a Tasmota-flashed SML smart meter IR read head
(e.g. Hichi "WiFi v2 Lesekopf") and Tasmota devices in general.

Like checkmk_tasmota_plug, this queries the device's open, documented local HTTP JSON
API (`http://<device>/cm?cmnd=Status 0`) -- no cloud account, no vendored third-party
library, no MQTT broker to run and keep alive. `Status 0` is Tasmota's "give me
everything" command: firmware, network, uptime, Wi-Fi, SML sensor data, all in one
response. This project previously also shipped a pure-MQTT based approach
(`tasmota_energymeter` v1.0.x: a `write_mqtt_to_file.py` cronjob plus an "individual
program call instead of agent" reading the resulting file) -- superseded by this special
agent, which needs nothing running on the Checkmk server or a broker in between: one
HTTP request per poll, straight to the device, confirmed against real Tasmota
13.4.0(tasmota32) firmware on an ESP32-C3 based Hichi read head.

It never fails loudly on a connection problem: an unreachable or misauthenticating
device still produces a `tasmota_energymeter_info` section marking itself unreachable, so
the corresponding Checkmk service goes CRIT with a clear reason instead of the host just
showing "no data received" (same defensive pattern as checkmk_tapo's agent_tapo and
checkmk_tasmota_plug's agent_tasmota_plug).

The functions below are split so the pure logic (extraction, URL building) is
unit-testable without a real device or the `cmk.*` runtime -- see
tests/test_agent_tasmota_energymeter.py.
"""

import argparse
import json
import sys
import urllib.parse
import urllib.request
from collections.abc import Mapping
from typing import Any

# Tasmota's web admin has exactly one fixed account name ("admin"); WebPassword only
# configures its password, there's no separate username setting to expose in the
# ruleset. Confirmed the same way for checkmk_tasmota_plug.
WEB_USERNAME = "admin"


def parse_arguments(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Checkmk special agent for a Tasmota based SML smart meter read head"
    )
    parser.add_argument("--host", required=True, help="Device IP address or hostname")
    parser.add_argument("--protocol", choices=("http", "https"), default="http")
    parser.add_argument("--port", type=int, default=80)
    password_group = parser.add_mutually_exclusive_group()
    password_group.add_argument(
        "--password-id", help="Password-store reference ('id:path'), resolved at runtime"
    )
    password_group.add_argument(
        "--password",
        help="Plaintext web admin (WebPassword) password. Only for manual CLI testing "
        "outside Checkmk -- Checkmk itself always passes --password-id when a password "
        "is configured. Leave unset entirely if the device has no WebPassword set.",
    )
    parser.add_argument("--timeout", type=int, default=10, help="Connection timeout in seconds")
    return parser.parse_args(argv)


def resolve_password(args: argparse.Namespace) -> str | None:
    if args.password_id:
        from cmk.password_store.v1_unstable import dereference_secret

        return dereference_secret(args.password_id).reveal()
    return args.password


def build_status_url(host: str, port: int, protocol: str, password: str | None) -> str:
    query = {"cmnd": "Status 0"}
    if password:
        # Tasmota's HTTP command API accepts credentials as plain query parameters when
        # the device's web admin password (WebPassword) is set -- confirmed against real
        # Tasmota firmware for checkmk_tasmota_plug and carried over here unchanged.
        query["user"] = WEB_USERNAME
        query["password"] = password
    return f"{protocol}://{host}:{port}/cm?{urllib.parse.urlencode(query)}"


def fetch_status(url: str, timeout: int) -> dict[str, Any]:
    with urllib.request.urlopen(url, timeout=timeout) as response:  # noqa: S310 (local device, not user input)
        return json.loads(response.read())


def relay_states(status: Mapping[str, Any]) -> dict[str, bool]:
    """Extract every POWER / POWER1 / POWER2 / ... key from StatusSTS as name->bool.

    An SML read head normally has no relay at all (StatusSTS reports none), but the same
    Tasmota firmware/template family is also used on plugs with one or more relays -- see
    checkmk_tasmota_plug's identical function. Kept here for free so a combined or
    relay-equipped device works without code changes; it simply yields an empty dict
    (and is not shown) on a plain read head.
    """
    sts = status.get("StatusSTS") or {}
    return {
        key: value == "ON"
        for key, value in sts.items()
        if key == "POWER" or (key.startswith("POWER") and key[len("POWER") :].isdigit())
    }


def build_info_payload(host: str, status: Mapping[str, Any]) -> dict[str, Any]:
    top = status.get("Status") or {}
    prm = status.get("StatusPRM") or {}
    fwr = status.get("StatusFWR") or {}
    net = status.get("StatusNET") or {}
    sts = status.get("StatusSTS") or {}
    wifi = sts.get("Wifi") or {}

    friendly_names = top.get("FriendlyName") or []
    device_name = top.get("DeviceName") or (friendly_names[0] if friendly_names else None)

    return {
        "reachable": True,
        "host": host,
        "device_name": device_name,
        "hostname": net.get("Hostname"),
        "relays": relay_states(status),
        "uptime_s": sts.get("UptimeSec"),
        "restart_reason": prm.get("RestartReason"),
        "firmware_version": fwr.get("Version"),
        "hardware": fwr.get("Hardware"),
        "ip_address": net.get("IPAddress"),
        "mac": net.get("Mac"),
        "wifi_ssid": wifi.get("SSId"),
        "wifi_rssi_pct": wifi.get("RSSI"),
        "wifi_signal_dbm": wifi.get("Signal"),
        # Device health, available for free in every Status 0 response but not covered
        # by either predecessor (tasmota_energymeter v1.0.x or tasmota_sml v1.1.x): free
        # heap is the classic ESP8266/ESP32 memory-leak indicator, LoadAvg is Tasmota's
        # own main-loop load figure -- both cheap early warnings of a device about to
        # crash/reboot on its own.
        "heap_free_kb": sts.get("Heap"),
        "loadavg": sts.get("LoadAvg"),
    }


def error_payload(host: str, error: BaseException) -> dict[str, Any]:
    return {"reachable": False, "host": host, "error": str(error)}


def run(args: argparse.Namespace) -> None:
    password = resolve_password(args)
    url = build_status_url(args.host, args.port, args.protocol, password)

    try:
        status = fetch_status(url, args.timeout)
    except Exception as exc:
        # Defensive by design, matching checkmk_tasmota_plug's agent_tasmota_plug: an
        # unreachable or misauthenticating device still yields a section (CRIT via
        # tasmota_energymeter_info), never a crash that leaves the host with no data at
        # all.
        print("<<<tasmota_energymeter_info:sep(0)>>>")
        print(json.dumps(error_payload(args.host, exc)))
        return

    print("<<<tasmota_energymeter_info:sep(0)>>>")
    print(json.dumps(build_info_payload(args.host, status)))

    # StatusSNS carries the SML sensor block (its own name varies, e.g. "SML" -- see
    # agent_based/tasmota_energymeter_sml.py's _find_sensor_block). Emitted as-is, one
    # section, one request -- the semantic old/new field-name mapping happens on the
    # check side so the raw agent output stays a faithful, inspectable copy of what the
    # device actually sent (useful for troubleshooting a device this extension doesn't
    # fully understand yet).
    status_sns = status.get("StatusSNS")
    if status_sns:
        print("<<<tasmota_energymeter_sml:sep(0)>>>")
        print(json.dumps(status_sns))


def main(argv: list[str] | None = None) -> int:
    args = parse_arguments(sys.argv[1:] if argv is None else argv)
    run(args)
    return 0


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