"""Read-only NAPALM driver for BMCs over IPMI 2.0, for boards that refuse or lack Redfish.

IPMI over LAN gives a password hash to anyone who names a valid user (CVE-2013-4786), so
this is the fallback, not the main path. It reads facts and inventory from the FRU,
sensors, the power state and the system event log, and switches the server's power
(chassis control). It does not reset the BMC; that goes through Redfish.
"""

from __future__ import annotations

from typing import Any

from napalm.base.exceptions import ConnectionException
from napalm_device_types import HealthMetricsMixin, ManagementControllerDriver
from napalm_device_types.models import POWER_ACTIONS
from pyghmi.exceptions import IpmiException
from pyghmi.ipmi.command import Command
from pyghmi.ipmi.private.session import Session

_JsonDict = dict[str, Any]

#: pyghmi's health bits (pyghmi.constants.Health).
_WARNING, _CRITICAL, _FAILED = 1, 2, 4
#: The only interface a BMC reports over IPMI: its own LAN channel.
_BMC = "bmc"
#: Role power actions -> pyghmi's chassis control. IPMI has no graceful restart, and pyghmi
#: offers no power cycle; neither is offered.
_CHASSIS_CONTROL = {
    "On": "on",
    "GracefulShutdown": "shutdown",
    "ForceOff": "off",
    "ForceRestart": "reset",
}


def _severity(health: int) -> str:
    if health & (_CRITICAL | _FAILED):
        return "Critical"
    if health & _WARNING:
        return "Warning"
    return "OK"


def _supply_present(reading: Any) -> bool:
    return "Present" in (getattr(reading, "states", None) or [])


class IpmiDriver(HealthMetricsMixin, ManagementControllerDriver):
    """NAPALM driver for BMCs over IPMI 2.0 (lanplus): reads, and the server's power.

    ``HealthMetricsMixin``: the BMC's own load, memory and interfaces over SNMP (UCD-MIB),
    where the BMC runs net-snmp. The managed server's sensors come from ``get_environment``.
    """

    platform = "ipmi"
    DRIVER_NAME = "ipmi"
    #: IPMI is UDP; discovery probes no UDP, so this driver has no fingerprint.
    default_port = 623

    def __init__(
        self,
        hostname: str,
        username: str,
        password: str,
        timeout: int = 60,
        optional_args: _JsonDict | None = None,
    ) -> None:
        self.hostname = hostname
        self.username = username
        self.password = password
        self.timeout = timeout
        self.optional_args: _JsonDict = optional_args or {}
        self.port = int(self.optional_args.get("port") or self.default_port)
        self._ipmi: Any = None

    # --- connection ----------------------------------------------------------------------

    def open(self) -> None:
        self._forget_session()
        try:
            self._ipmi = Command(
                bmc=self.hostname,
                userid=self.username,
                password=self.password,
                port=self.port,
            )
        except Exception as exc:
            raise ConnectionException(f"Cannot connect to IPMI at {self.hostname}: {exc}") \
                from exc

    def close(self) -> None:
        ipmi, self._ipmi = self._ipmi, None
        if ipmi is not None:
            try:
                ipmi.ipmi_session.logout()
            except Exception:  # a session left open times out on the BMC
                pass
        self._forget_session()

    def _forget_session(self) -> None:
        """Drop this BMC's session from pyghmi's cache of sessions being set up.

        pyghmi files a new session there under ``(bmc, userid, password, port, kg)`` with
        the credentials as str, and only ever deletes it under the credentials as bytes, so
        the entry stays. The next ``Command`` for the same BMC in this worker process then
        got the old, logged-out session back and failed with ``'<' not supported between
        instances of 'float' and 'NoneType'``.
        """
        key = (self.hostname, self.username, self.password, self.port, None)
        Session.initting_sessions.pop(key, None)

    def is_alive(self) -> _JsonDict:
        return {"is_alive": self._ipmi is not None}

    def _cmd(self) -> Any:
        if self._ipmi is None:
            raise ConnectionException(f"{self.hostname}: not connected, call open() first")
        return self._ipmi

    # --- reads ---------------------------------------------------------------------------

    def _system(self) -> _JsonDict:
        """The "System" FRU record: manufacturer, product, serial number, UUID."""
        for name, record in self._cmd().get_inventory():
            if name == "System":
                return record or {}
        return {}

    def _bmc_version(self) -> str:
        for name, info in self._cmd().get_firmware():
            if name == "BMC Version":
                return str(info.get("version") or "")
        return ""

    def _sensors(self) -> list[Any]:
        return list(self._cmd().get_sensor_data())

    # --- NAPALM getters ------------------------------------------------------------------

    def get_facts(self) -> _JsonDict:
        system = self._system()
        return {
            "uptime": -1.0,  # IPMI has no uptime
            "vendor": system.get("Manufacturer") or "",
            "model": system.get("Product name") or system.get("Model") or "",
            "hostname": self.hostname,
            "fqdn": self.hostname,
            "os_version": self._bmc_version(),
            "serial_number": system.get("Serial Number") or "",
            "interface_list": [_BMC],
        }

    def get_interfaces(self) -> _JsonDict:
        net = self._cmd().get_net_configuration()
        source = net.get("ipv4_configuration") or ""
        return {
            _BMC: {
                "is_up": True,
                "is_enabled": True,
                "description": f"BMC LAN ({source})" if source else "BMC LAN",
                "last_flapped": -1.0,
                "speed": -1.0,
                "mtu": -1,
                "mac_address": (net.get("mac_address") or "").lower(),
            }
        }

    def get_interfaces_ip(self) -> _JsonDict:
        address = self._cmd().get_net_configuration().get("ipv4_address") or ""
        if "/" not in address:
            return {}
        ip, prefix = address.split("/", 1)
        return {_BMC: {"ipv4": {ip: {"prefix_length": int(prefix)}}}}

    def get_environment(self) -> _JsonDict:
        """Fans, temperatures and present power supplies from the sensor records; ``cpu`` and
        ``memory`` are empty, since a BMC does not know the host OS's load."""
        fans: _JsonDict = {}
        temperature: _JsonDict = {}
        power: _JsonDict = {}
        for reading in self._sensors():
            health = int(getattr(reading, "health", 0) or 0)
            value = getattr(reading, "value", None)
            if reading.type == "Temperature" and value is not None:
                temperature[reading.name] = {
                    "temperature": float(value),
                    "is_alert": bool(health),
                    "is_critical": bool(health & (_CRITICAL | _FAILED)),
                }
            elif reading.type == "Fan" and value is not None:
                fans[reading.name] = {"status": value > 0 and not health}
            elif reading.type == "Power Supply" and _supply_present(reading):
                power[reading.name] = {"status": not health, "capacity": -1.0, "output": -1.0}
        return {"fans": fans, "temperature": temperature, "power": power, "cpu": {}, "memory": {}}

    # --- role methods --------------------------------------------------------------------

    def _power_state(self) -> str:
        return str(self._cmd().get_power().get("powerstate") or "").capitalize()

    @staticmethod
    def _accepted_actions(state: str) -> list[str]:
        """The power actions that make sense in ``state``; IPMI advertises none itself."""
        if state == "Off":
            return ["On"]
        if state == "On":
            return [a for a in POWER_ACTIONS if a in _CHASSIS_CONTROL and a != "On"]
        return []

    def get_power_state(self) -> _JsonDict:
        state = self._power_state()
        return {"state": state, "actions": self._accepted_actions(state)}

    def set_power_state(self, action: str) -> None:
        if action not in POWER_ACTIONS:
            raise ValueError(f"{action!r} is not one of {', '.join(POWER_ACTIONS)}")
        state = self._power_state()
        accepted = self._accepted_actions(state)
        if action not in accepted:
            raise RuntimeError(
                f"{self.hostname}: {action} not accepted while the server is "
                f"{state or 'in an unknown state'} (accepted: {', '.join(accepted) or 'none'})")
        try:
            self._cmd().set_power(_CHASSIS_CONTROL[action], wait=False)
        except IpmiException as exc:
            raise RuntimeError(f"{self.hostname}: {action} refused: {exc}") from exc

    def get_hardware_inventory(self) -> _JsonDict:
        """What the FRU and the sensor records tell; IPMI lists no CPUs, DIMMs or drives."""
        system = self._system()
        supplies = [
            {
                "slot": reading.name.removesuffix(" Status"),
                "model": "",
                "serial_number": "",
                "capacity_watts": None,
                "status": _severity(int(reading.health or 0))
                if _supply_present(reading) else "Absent",
            }
            for reading in self._sensors()
            if reading.type == "Power Supply"
        ]
        return {
            "manufacturer": system.get("Manufacturer") or "",
            "model": system.get("Product name") or system.get("Model") or "",
            "serial_number": system.get("Serial Number") or "",
            "system_uuid": str(system.get("UUID") or "").lower(),
            "cpus": [],
            "memory": [],
            "drives": [],
            "power_supplies": supplies,
            "nics": [],
        }

    def get_event_log(self, limit: int = 500) -> list[_JsonDict]:
        """The system event log, newest first."""
        entries = []
        for event in self._cmd().get_event_log():
            component = event.get("component") or ""
            text = event.get("event") or ""
            timecode = event.get("timecode")
            entries.append({
                "id": str(event.get("record_id", "")),
                "timestamp": float(timecode) if timecode else None,
                "severity": _severity(int(event.get("severity") or 0)),
                "message": f"{component}: {text}" if component and text else component or text,
            })
        entries.sort(key=lambda e: (e["timestamp"] is not None, e["timestamp"] or 0.0),
                     reverse=True)
        return entries[:limit]

    def get_firmware_inventory(self) -> list[_JsonDict]:
        version = self._bmc_version()
        return [{"name": "BMC", "version": version, "updateable": None}] if version else []