temperature_monitor.py

#!/usr/bin/env python3
"""Record, analyze, plot, and export hardware temperature measurements.

Usage: python3.14 temperature_monitor.py [GLOBAL_OPTIONS] COMMAND [OPTIONS]
Requires psutil and matplotlib; the Windows LHM backend also needs pythonnet.
"""

from __future__ import annotations

import argparse
import csv
import json
import sqlite3
import statistics
import subprocess
import sys
import threading
import time
from abc import ABC, abstractmethod
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, ClassVar

import matplotlib.dates as mdates
import psutil
from matplotlib.backends._backend_tk import NavigationToolbar2Tk
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
from matplotlib.figure import Figure

try:
    import tkinter as tk
    from tkinter import messagebox, ttk
except ImportError as exc:  # pragma: no cover - tkinter is part of stdlib on normal installs
    raise SystemExit("Tkinter is required for the 'live' mode.") from exc


APP_NAME = "temperature-monitor"
DEFAULT_DB_PATH = Path.home() / ".temperature_monitor.sqlite3"


@dataclass(frozen=True, slots=True)
class SensorDescriptor:
    key: str
    name: str
    component: str
    unit: str
    source: str
    metadata: dict[str, Any]


@dataclass(frozen=True, slots=True)
class SensorReading:
    sensor_key: str
    captured_at: datetime
    value: float


@dataclass(frozen=True, slots=True)
class SessionRecord:
    id: int
    name: str | None
    backend_name: str
    started_at: str
    ended_at: str | None


def utc_now() -> datetime:
    return datetime.now(UTC)


def utc_now_text() -> str:
    return utc_now().isoformat()


def parse_timestamp(value: str) -> datetime:
    return datetime.fromisoformat(value)


def read_psutil_temperatures() -> dict[str, Any]:
    getter = getattr(psutil, "sensors_temperatures", None)
    if getter is None:
        raise NotImplementedError(
            "psutil does not expose sensors_temperatures() in this environment"
        )
    return getter()


def format_session_label(session: SessionRecord) -> str:
    title = session.name or f"session-{session.id}"
    return f"{session.id}: {title} [{session.started_at}]"


def quantiles(values: list[float]) -> tuple[float, float, float]:
    ordered = sorted(values)
    if len(ordered) == 1:
        return ordered[0], ordered[0], ordered[0]
    quartiles = statistics.quantiles(ordered, n=4, method="inclusive")
    return quartiles[0], quartiles[1], quartiles[2]


class MetricBackend(ABC):
    backend_name: str

    @abstractmethod
    def check_support(self) -> tuple[bool, str | None]:
        raise NotImplementedError

    @abstractmethod
    def discover_sensors(self) -> list[SensorDescriptor]:
        raise NotImplementedError

    @abstractmethod
    def sample(self) -> list[SensorReading]:
        raise NotImplementedError

    def close(self) -> None:
        return None


class PsutilTemperatureBackend(MetricBackend):
    backend_name = "psutil"

    def check_support(self) -> tuple[bool, str | None]:
        try:
            sensors = read_psutil_temperatures()
        except (AttributeError, NotImplementedError) as exc:
            return False, f"psutil sensors API is unavailable: {exc}"
        if not sensors:
            return False, "psutil did not return any temperature sensors on this system"
        return True, None

    def discover_sensors(self) -> list[SensorDescriptor]:
        sensors = read_psutil_temperatures()
        descriptors: list[SensorDescriptor] = []
        for chip_name, entries in sensors.items():
            for index, entry in enumerate(entries):
                label = entry.label or f"{chip_name}-{index}"
                sensor_key = f"{chip_name}:{label}:{index}"
                descriptors.append(
                    SensorDescriptor(
                        key=sensor_key,
                        name=label,
                        component=chip_name,
                        unit="C",
                        source=self.backend_name,
                        metadata={
                            "chip": chip_name,
                            "index": index,
                            "high": entry.high,
                            "critical": entry.critical,
                        },
                    )
                )
        return descriptors

    def sample(self) -> list[SensorReading]:
        captured_at = utc_now()
        readings: list[SensorReading] = []
        for chip_name, entries in read_psutil_temperatures().items():
            for index, entry in enumerate(entries):
                if entry.current is None:
                    continue
                label = entry.label or f"{chip_name}-{index}"
                sensor_key = f"{chip_name}:{label}:{index}"
                readings.append(
                    SensorReading(
                        sensor_key=sensor_key,
                        captured_at=captured_at,
                        value=float(entry.current),
                    )
                )
        return readings


class LibreHardwareMonitorBackend(MetricBackend):
    backend_name = "librehardwaremonitor"
    REQUIRED_SIBLING_ASSEMBLIES: ClassVar[set[str]] = {
        "BlackSharp.Core",
        "DiskInfoToolkit",
        "HidSharp",
        "RAMSPDToolkit-NDD",
        "System.Memory",
        "System.Runtime.CompilerServices.Unsafe",
    }

    def __init__(self, dll_path: Path | None = None) -> None:
        self._dll_path = dll_path
        self._computer: Any | None = None
        self._assembly: Any | None = None

    def check_support(self) -> tuple[bool, str | None]:
        if not sys.platform.startswith("win"):
            return False, "LibreHardwareMonitor backend is supported only on Windows"
        try:
            self._ensure_initialized()
        except RuntimeError as exc:
            return False, str(exc)
        return True, None

    def discover_sensors(self) -> list[SensorDescriptor]:
        self._ensure_initialized()
        descriptors: list[SensorDescriptor] = []
        for hardware_name, hardware_type, sensor in self._iter_temperature_sensors():
            sensor_key = self._sensor_key(hardware_name, sensor.Name, sensor.Identifier)
            descriptors.append(
                SensorDescriptor(
                    key=sensor_key,
                    name=str(sensor.Name),
                    component=f"{hardware_type}:{hardware_name}",
                    unit="C",
                    source=self.backend_name,
                    metadata={
                        "identifier": str(sensor.Identifier),
                        "hardware_name": hardware_name,
                        "hardware_type": hardware_type,
                    },
                )
            )
        return descriptors

    def sample(self) -> list[SensorReading]:
        self._ensure_initialized()
        captured_at = utc_now()
        readings: list[SensorReading] = []
        for hardware_name, _, sensor in self._iter_temperature_sensors():
            if sensor.Value is None:
                continue
            sensor_key = self._sensor_key(hardware_name, sensor.Name, sensor.Identifier)
            readings.append(
                SensorReading(
                    sensor_key=sensor_key,
                    captured_at=captured_at,
                    value=float(sensor.Value),
                )
            )
        return readings

    def close(self) -> None:
        if self._computer is not None:
            self._computer.Close()
            self._computer = None
            self._assembly = None

    def _ensure_initialized(self) -> None:
        if self._computer is not None:
            return
        dll_path = self._resolve_dll_path()
        try:
            import clr  # type: ignore[import-not-found]
        except ImportError as exc:
            raise RuntimeError(
                "pythonnet is not installed. Install it first, then provide LibreHardwareMonitorLib.dll."
            ) from exc

        try:
            clr.AddReference(str(dll_path))  # type: ignore[attr-defined]
            from System import Activator  # type: ignore[import-not-found]
            from System.Reflection import Assembly  # type: ignore[import-not-found]

            assembly = Assembly.LoadFile(str(dll_path))
            target_framework = self._read_target_framework(assembly)
            if target_framework and target_framework.startswith(".NETCoreApp"):
                raise RuntimeError(
                    "The detected LibreHardwareMonitorLib.dll targets "
                    f"{target_framework}, but pythonnet is running on .NET Framework. "
                    "Use the LibreHardwareMonitor-net472 build instead of the .NET/NET.10 build."
                )
            missing_dependencies = self._missing_sibling_assemblies(assembly, dll_path.parent)
            if missing_dependencies:
                missing_list = ", ".join(f"{name}.dll" for name in missing_dependencies)
                raise RuntimeError(
                    "LibreHardwareMonitor dependencies are missing next to the main DLL: "
                    f"{missing_list}. Copy the full contents of LibreHardwareMonitor-net472.zip, "
                    "not just LibreHardwareMonitorLib.dll."
                )
            computer_type = assembly.GetType("LibreHardwareMonitor.Hardware.Computer")
            if computer_type is None:
                raise RuntimeError(
                    "Type LibreHardwareMonitor.Hardware.Computer was not found in the DLL."
                )
            computer = Activator.CreateInstance(computer_type)
        except Exception as exc:  # pragma: no cover - depends on external DLL/runtime
            raise RuntimeError(
                f"Failed to initialize LibreHardwareMonitor from {dll_path}. Original error: {exc}"
            ) from exc

        computer.IsBatteryEnabled = True
        computer.IsControllerEnabled = True
        computer.IsCpuEnabled = True
        computer.IsGpuEnabled = True
        computer.IsMemoryEnabled = True
        computer.IsMotherboardEnabled = True
        computer.IsNetworkEnabled = True
        computer.IsPsuEnabled = True
        computer.IsStorageEnabled = True
        if hasattr(computer, "IsPowerMonitorEnabled"):
            computer.IsPowerMonitorEnabled = True
        computer.Open()

        self._computer = computer
        self._assembly = assembly

    def _resolve_dll_path(self) -> Path:
        candidates: list[Path] = []
        if self._dll_path is not None:
            candidates.append(normalize_lhm_path(self._dll_path))
        env_path = Path.home() / "LibreHardwareMonitor" / "LibreHardwareMonitorLib.dll"
        configured_lhm_path = get_configured_lhm_path()
        if configured_lhm_path is not None:
            candidates.append(configured_lhm_path)
        candidates.extend(
            [
                Path.cwd() / "LibreHardwareMonitorLib.dll",
                Path(__file__).resolve().parent / "liblhm" / "LibreHardwareMonitorLib.dll",
                Path(__file__).with_name("LibreHardwareMonitorLib.dll"),
                env_path,
            ]
        )
        for candidate in candidates:
            if candidate.is_file():
                return candidate
        looked = ", ".join(str(path) for path in candidates)
        raise RuntimeError(
            "LibreHardwareMonitorLib.dll was not found. "
            f"Looked in: {looked}. Set TEMPMON_LHM_DLL or use --windows-lhm-dll."
        )

    def _iter_temperature_sensors(self) -> list[tuple[str, str, Any]]:
        assert self._computer is not None
        sensors: list[tuple[str, str, Any]] = []
        for hardware in self._computer.Hardware:
            sensors.extend(self._collect_from_hardware(hardware))
        return sensors

    def _collect_from_hardware(self, hardware: Any) -> list[tuple[str, str, Any]]:
        hardware.Update()
        hardware_name = str(hardware.Name)
        hardware_type = str(hardware.HardwareType)
        found = [
            (hardware_name, hardware_type, sensor)
            for sensor in hardware.Sensors
            if str(sensor.SensorType) == "Temperature"
        ]
        for sub_hardware in hardware.SubHardware:
            found.extend(self._collect_from_hardware(sub_hardware))
        return found

    @staticmethod
    def _sensor_key(hardware_name: str, sensor_name: Any, sensor_identifier: Any) -> str:
        return f"{hardware_name}:{sensor_name}:{sensor_identifier}"

    @staticmethod
    def _read_target_framework(assembly: Any) -> str | None:
        for attr in assembly.GetCustomAttributesData():
            if attr.AttributeType.FullName == "System.Runtime.Versioning.TargetFrameworkAttribute":
                arguments = list(attr.ConstructorArguments)
                if arguments:
                    return str(arguments[0].Value)
        return None

    @classmethod
    def _missing_sibling_assemblies(cls, assembly: Any, dll_dir: Path) -> list[str]:
        missing: list[str] = []
        for reference in assembly.GetReferencedAssemblies():
            name = str(reference.Name)
            if name not in cls.REQUIRED_SIBLING_ASSEMBLIES:
                continue
            if not (dll_dir / f"{name}.dll").is_file():
                missing.append(name)
        return missing


def os_environ() -> Mapping[str, str]:
    import os

    return os.environ


def normalize_lhm_path(path: Path) -> Path:
    if path.is_dir():
        return path / "LibreHardwareMonitorLib.dll"
    return path


def get_configured_lhm_path() -> Path | None:
    value = os_environ().get("TEMPMON_LHM_DLL")
    if not value:
        return None
    return normalize_lhm_path(Path(value))


def is_windows_admin() -> bool:
    if not sys.platform.startswith("win"):
        return False
    try:
        import ctypes

        windll = vars(ctypes)["windll"]
        return bool(windll.shell32.IsUserAnAdmin())
    except Exception:
        return False


def should_auto_relaunch_as_admin(args: argparse.Namespace) -> bool:
    if not sys.platform.startswith("win"):
        return False
    if args.command not in {"watch", "live"}:
        return False
    if args.backend == "psutil":
        return False
    if getattr(args, "_elevated_relaunch", False):
        return False
    if getattr(args, "no_admin_relaunch", False):
        return False
    return not is_windows_admin()


def relaunch_as_admin() -> int:
    import ctypes

    script_path = Path(sys.argv[0]).resolve()
    original_args = list(sys.argv[1:])
    command_index = next(
        (
            index
            for index, value in enumerate(original_args)
            if value in {"watch", "live", "analyze", "stats", "export"}
        ),
        len(original_args),
    )
    pre_command_args = original_args[:command_index]
    post_command_args = original_args[command_index:]
    forwarded_args = [str(script_path), *pre_command_args]
    configured_lhm_path = get_configured_lhm_path()
    if configured_lhm_path is not None and "--windows-lhm-dll" not in original_args:
        forwarded_args.extend(["--windows-lhm-dll", str(configured_lhm_path)])
    forwarded_args.extend(["--_elevated-relaunch", *post_command_args])
    parameters = subprocess.list2cmdline(forwarded_args)
    windll = vars(ctypes)["windll"]
    result = windll.shell32.ShellExecuteW(
        None,
        "runas",
        sys.executable,
        parameters,
        str(Path.cwd()),
        1,
    )
    if result <= 32:
        raise RuntimeError(
            "LibreHardwareMonitor needs administrator rights on this machine to read temperature/clock sensors. "
            "UAC elevation was not completed."
        )
    return 0


def show_windows_error_dialog(title: str, message: str) -> None:
    if not sys.platform.startswith("win"):
        return
    try:
        import ctypes

        windll = vars(ctypes)["windll"]
        windll.user32.MessageBoxW(None, message, title, 0x10)
    except Exception:
        return


class BackendFactory:
    def __init__(self, windows_lhm_dll: Path | None = None) -> None:
        self._windows_lhm_dll = windows_lhm_dll

    def create(self, requested: str) -> MetricBackend:
        backends = self._candidate_backends(requested)
        errors: list[str] = []
        for backend in backends:
            supported, reason = backend.check_support()
            if supported:
                return backend
            errors.append(f"{backend.backend_name}: {reason}")
            backend.close()
        raise RuntimeError("No supported temperature backend found.\n" + "\n".join(errors))

    def _candidate_backends(self, requested: str) -> list[MetricBackend]:
        if requested == "psutil":
            return [PsutilTemperatureBackend()]
        if requested == "librehardwaremonitor":
            return [LibreHardwareMonitorBackend(self._windows_lhm_dll)]
        if sys.platform.startswith("win"):
            return [
                LibreHardwareMonitorBackend(self._windows_lhm_dll),
                PsutilTemperatureBackend(),
            ]
        return [PsutilTemperatureBackend()]


class MeasurementStore:
    def __init__(self, db_path: Path) -> None:
        self.db_path = db_path
        self.connection = sqlite3.connect(self.db_path)
        self.connection.row_factory = sqlite3.Row
        self._init_schema()

    def close(self) -> None:
        self.connection.close()

    def _init_schema(self) -> None:
        self.connection.executescript(
            """
            PRAGMA journal_mode = WAL;

            CREATE TABLE IF NOT EXISTS sessions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT,
                backend_name TEXT NOT NULL,
                started_at TEXT NOT NULL,
                ended_at TEXT,
                metadata_json TEXT NOT NULL DEFAULT '{}'
            );

            CREATE TABLE IF NOT EXISTS sensors (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id INTEGER NOT NULL,
                sensor_key TEXT NOT NULL,
                name TEXT NOT NULL,
                component TEXT NOT NULL,
                unit TEXT NOT NULL,
                source TEXT NOT NULL,
                metadata_json TEXT NOT NULL DEFAULT '{}',
                UNIQUE(session_id, sensor_key),
                FOREIGN KEY(session_id) REFERENCES sessions(id)
            );

            CREATE TABLE IF NOT EXISTS measurements (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id INTEGER NOT NULL,
                sensor_id INTEGER NOT NULL,
                captured_at TEXT NOT NULL,
                value REAL NOT NULL,
                FOREIGN KEY(session_id) REFERENCES sessions(id),
                FOREIGN KEY(sensor_id) REFERENCES sensors(id)
            );

            CREATE INDEX IF NOT EXISTS idx_sessions_started_at ON sessions(started_at DESC);
            CREATE INDEX IF NOT EXISTS idx_sensors_session ON sensors(session_id);
            CREATE INDEX IF NOT EXISTS idx_measurements_session_time ON measurements(session_id, captured_at);
            CREATE INDEX IF NOT EXISTS idx_measurements_sensor_time ON measurements(sensor_id, captured_at);
            """
        )
        self.connection.commit()

    def create_session(
        self, backend_name: str, name: str | None, metadata: dict[str, Any]
    ) -> SessionRecord:
        started_at = utc_now_text()
        cursor = self.connection.execute(
            """
            INSERT INTO sessions (name, backend_name, started_at, metadata_json)
            VALUES (?, ?, ?, ?)
            """,
            (name, backend_name, started_at, json.dumps(metadata, ensure_ascii=False)),
        )
        self.connection.commit()
        session_id = cursor.lastrowid
        if session_id is None:
            raise RuntimeError("Failed to create a session row.")
        return SessionRecord(
            id=int(session_id),
            name=name,
            backend_name=backend_name,
            started_at=started_at,
            ended_at=None,
        )

    def mark_session_closed(self, session_id: int) -> None:
        self.connection.execute(
            "UPDATE sessions SET ended_at = ? WHERE id = ?",
            (utc_now_text(), session_id),
        )
        self.connection.commit()

    def clear_session_closed(self, session_id: int) -> None:
        self.connection.execute("UPDATE sessions SET ended_at = NULL WHERE id = ?", (session_id,))
        self.connection.commit()

    def resolve_session(self, session_spec: str | None) -> SessionRecord:
        if session_spec is None:
            row = self.connection.execute(
                """
                SELECT id, name, backend_name, started_at, ended_at
                FROM sessions
                ORDER BY started_at DESC
                LIMIT 1
                """
            ).fetchone()
            if row is None:
                raise RuntimeError("There are no sessions yet.")
            return self._row_to_session(row)

        if session_spec.isdigit():
            row = self.connection.execute(
                """
                SELECT id, name, backend_name, started_at, ended_at
                FROM sessions
                WHERE id = ?
                """,
                (int(session_spec),),
            ).fetchone()
            if row is not None:
                return self._row_to_session(row)

        row = self.connection.execute(
            """
            SELECT id, name, backend_name, started_at, ended_at
            FROM sessions
            WHERE name = ?
            ORDER BY started_at DESC
            LIMIT 1
            """,
            (session_spec,),
        ).fetchone()
        if row is None:
            raise RuntimeError(f"Session '{session_spec}' was not found.")
        return self._row_to_session(row)

    def list_sessions(self) -> list[SessionRecord]:
        rows = self.connection.execute(
            """
            SELECT id, name, backend_name, started_at, ended_at
            FROM sessions
            ORDER BY started_at DESC
            """
        ).fetchall()
        return [self._row_to_session(row) for row in rows]

    def register_sensors(self, session_id: int, sensors: list[SensorDescriptor]) -> None:
        rows = [
            (
                session_id,
                sensor.key,
                sensor.name,
                sensor.component,
                sensor.unit,
                sensor.source,
                json.dumps(sensor.metadata, ensure_ascii=False),
            )
            for sensor in sensors
        ]
        self.connection.executemany(
            """
            INSERT INTO sensors (session_id, sensor_key, name, component, unit, source, metadata_json)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(session_id, sensor_key) DO UPDATE SET
                name = excluded.name,
                component = excluded.component,
                unit = excluded.unit,
                source = excluded.source,
                metadata_json = excluded.metadata_json
            """,
            rows,
        )
        self.connection.commit()

    def insert_measurements(self, session_id: int, readings: list[SensorReading]) -> int:
        if not readings:
            return 0
        sensor_map = self._sensor_ids(session_id)
        rows = []
        for reading in readings:
            sensor_id = sensor_map.get(reading.sensor_key)
            if sensor_id is None:
                continue
            rows.append(
                (
                    session_id,
                    sensor_id,
                    reading.captured_at.isoformat(),
                    reading.value,
                )
            )
        if not rows:
            return 0
        self.connection.executemany(
            """
            INSERT INTO measurements (session_id, sensor_id, captured_at, value)
            VALUES (?, ?, ?, ?)
            """,
            rows,
        )
        self.connection.commit()
        return len(rows)

    def get_sensor_rows(self, session_id: int) -> list[sqlite3.Row]:
        return self.connection.execute(
            """
            SELECT id, sensor_key, name, component, unit, source, metadata_json
            FROM sensors
            WHERE session_id = ?
            ORDER BY component, name
            """,
            (session_id,),
        ).fetchall()

    def get_measurements_for_session(self, session_id: int) -> list[sqlite3.Row]:
        return self.connection.execute(
            """
            SELECT
                m.captured_at,
                m.value,
                s.sensor_key,
                s.name,
                s.component,
                s.unit
            FROM measurements AS m
            JOIN sensors AS s ON s.id = m.sensor_id
            WHERE m.session_id = ?
            ORDER BY m.captured_at ASC
            """,
            (session_id,),
        ).fetchall()

    def get_measurements_for_sensor(self, session_id: int, sensor_key: str) -> list[sqlite3.Row]:
        return self.connection.execute(
            """
            SELECT
                m.captured_at,
                m.value
            FROM measurements AS m
            JOIN sensors AS s ON s.id = m.sensor_id
            WHERE m.session_id = ? AND s.sensor_key = ?
            ORDER BY m.captured_at ASC
            """,
            (session_id, sensor_key),
        ).fetchall()

    def _sensor_ids(self, session_id: int) -> dict[str, int]:
        rows = self.connection.execute(
            "SELECT sensor_key, id FROM sensors WHERE session_id = ?",
            (session_id,),
        ).fetchall()
        return {str(row["sensor_key"]): int(row["id"]) for row in rows}

    @staticmethod
    def _row_to_session(row: sqlite3.Row) -> SessionRecord:
        session_id = row["id"]
        if session_id is None:
            raise RuntimeError("Session row does not contain an id.")
        return SessionRecord(
            id=int(session_id),
            name=row["name"],
            backend_name=str(row["backend_name"]),
            started_at=str(row["started_at"]),
            ended_at=row["ended_at"],
        )


class Recorder:
    def __init__(
        self, store: MeasurementStore, backend: MetricBackend, session: SessionRecord
    ) -> None:
        self.store = store
        self.backend = backend
        self.session = session

    def prepare(self) -> list[SensorDescriptor]:
        sensors = self.backend.discover_sensors()
        if not sensors:
            raise RuntimeError("No temperature sensors were discovered.")
        self.store.register_sensors(self.session.id, sensors)
        return sensors

    def capture_once(self) -> int:
        readings = self.backend.sample()
        inserted = self.store.insert_measurements(self.session.id, readings)
        if inserted == 0 and readings:
            sensors = self.backend.discover_sensors()
            if sensors:
                self.store.register_sensors(self.session.id, sensors)
                inserted = self.store.insert_measurements(self.session.id, readings)
        return inserted


def collect_initial_readings(
    backend: MetricBackend,
    attempts: int = 5,
    delay_seconds: float = 0.5,
) -> list[SensorReading]:
    for attempt in range(attempts):
        readings = backend.sample()
        if readings:
            return readings
        if attempt < attempts - 1:
            time.sleep(delay_seconds)
    return []


def describe_sensor_readability(
    backend: MetricBackend,
    sensors: list[SensorDescriptor],
    readings: list[SensorReading],
) -> str:
    if not sensors:
        return "No temperature sensors were discovered yet."
    if readings:
        return f"Discovered {len(sensors)} sensors and collected {len(readings)} readings."
    if backend.backend_name == "librehardwaremonitor" and not is_windows_admin():
        return (
            f"Discovered {len(sensors)} sensors via {backend.backend_name}, "
            "but readable temperature values are not available in this non-admin session. "
            "Relaunch elevated to unlock MSR-backed CPU temperature sensors."
        )
    return (
        f"Discovered {len(sensors)} sensors via {backend.backend_name}, "
        "but none of them currently expose readable temperature values."
    )


def bootstrap_new_session(
    store: MeasurementStore,
    backend: MetricBackend,
    session_name: str | None,
    metadata: dict[str, Any],
) -> tuple[SessionRecord, Recorder, list[SensorDescriptor], int]:
    sensors = backend.discover_sensors()
    if not sensors:
        raise RuntimeError("No temperature sensors were discovered.")

    initial_readings = collect_initial_readings(backend)
    if not initial_readings:
        raise RuntimeError(
            "Temperature sensors were discovered, but no readable temperature values were returned."
        )

    session = store.create_session(
        backend_name=backend.backend_name,
        name=session_name,
        metadata=metadata,
    )
    store.register_sensors(session.id, sensors)
    inserted = store.insert_measurements(session.id, initial_readings)
    if inserted == 0:
        raise RuntimeError(
            "Initial temperature readings were collected, but they did not match the registered sensors."
        )

    return session, Recorder(store, backend, session), sensors, inserted


def wait_for_bootstrap_session(
    store: MeasurementStore,
    backend: MetricBackend,
    session_name: str | None,
    metadata: dict[str, Any],
    interval_seconds: float,
) -> tuple[SessionRecord, Recorder, list[SensorDescriptor], int]:
    while True:
        sensors = backend.discover_sensors()
        readings = collect_initial_readings(backend, attempts=1, delay_seconds=0.0)
        if sensors and readings:
            session = store.create_session(
                backend_name=backend.backend_name,
                name=session_name,
                metadata=metadata,
            )
            store.register_sensors(session.id, sensors)
            inserted = store.insert_measurements(session.id, readings)
            if inserted == 0:
                raise RuntimeError(
                    "Initial temperature readings were collected, but they did not match the registered sensors."
                )
            return session, Recorder(store, backend, session), sensors, inserted

        print(describe_sensor_readability(backend, sensors, readings))
        time.sleep(interval_seconds)


class WatchWorker:
    def __init__(
        self,
        db_path: Path,
        factory: BackendFactory,
        backend_name: str,
        session_id: int | None,
        interval_seconds: float,
        session_name: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        self.db_path = db_path
        self.factory = factory
        self.backend_name = backend_name
        self.session_id = session_id
        self.interval_seconds = interval_seconds
        self.session_name = session_name
        self.metadata = metadata or {}
        self._stop_event = threading.Event()
        thread_suffix = session_id if session_id is not None else "pending"
        self._thread = threading.Thread(target=self._run, name=f"watch-session-{thread_suffix}")
        self._status_lock = threading.Lock()
        self._status = "Collector is starting..."
        self._error: str | None = None

    def start(self) -> None:
        self._thread.start()

    def stop(self) -> None:
        self._stop_event.set()
        self._thread.join(timeout=max(1.0, self.interval_seconds + 1.0))

    def status_text(self) -> str:
        with self._status_lock:
            return self._status

    def error_text(self) -> str | None:
        with self._status_lock:
            return self._error

    def _set_status(self, status: str, error: str | None = None) -> None:
        with self._status_lock:
            self._status = status
            self._error = error

    def _run(self) -> None:
        store = MeasurementStore(self.db_path)
        backend: MetricBackend | None = None
        active_session_id: int | None = None
        try:
            backend = self.factory.create(self.backend_name)
            if self.session_id is None:
                session, recorder, sensors, inserted = self._wait_for_new_session(store, backend)
                self.session_id = session.id
                active_session_id = session.id
                self._set_status(
                    f"Collecting into session {session.id} with {len(sensors)} sensors every {self.interval_seconds:g}s. "
                    f"Initial insert: {inserted} readings."
                )
            else:
                session = store.resolve_session(str(self.session_id))
                active_session_id = session.id
                recorder = Recorder(store, backend, session)
                sensors = recorder.prepare()
                self._set_status(
                    f"Collecting into session {session.id} with {len(sensors)} sensors every {self.interval_seconds:g}s."
                )

            while not self._stop_event.is_set():
                inserted = recorder.capture_once()
                self._set_status(
                    f"Collecting into session {session.id}; last insert: {inserted} readings."
                )
                if self._stop_event.wait(self.interval_seconds):
                    break
        except Exception as exc:
            session_label = self.session_id if self.session_id is not None else "pending session"
            self._set_status(f"Collector error in session {session_label}: {exc}", error=str(exc))
        finally:
            try:
                if active_session_id is not None:
                    store.mark_session_closed(active_session_id)
            finally:
                if backend is not None:
                    backend.close()
                store.close()

    def _wait_for_new_session(
        self,
        store: MeasurementStore,
        backend: MetricBackend,
    ) -> tuple[SessionRecord, Recorder, list[SensorDescriptor], int]:
        while not self._stop_event.is_set():
            sensors = backend.discover_sensors()
            readings = collect_initial_readings(backend, attempts=1, delay_seconds=0.0)
            if sensors and readings:
                session = store.create_session(
                    backend_name=backend.backend_name,
                    name=self.session_name,
                    metadata=self.metadata,
                )
                store.register_sensors(session.id, sensors)
                inserted = store.insert_measurements(session.id, readings)
                if inserted == 0:
                    raise RuntimeError(
                        "Initial temperature readings were collected, but they did not match the registered sensors."
                    )
                return session, Recorder(store, backend, session), sensors, inserted

            self._set_status(describe_sensor_readability(backend, sensors, readings))
            if self._stop_event.wait(self.interval_seconds):
                raise RuntimeError(
                    "Collector was stopped before any readable temperature values appeared."
                )

        raise RuntimeError("Collector was stopped before initialization.")


class LiveMonitorWindow:
    def __init__(
        self,
        store: MeasurementStore,
        initial_session_id: int | None,
        worker: WatchWorker | None,
    ) -> None:
        self.store = store
        self.worker = worker
        self.current_collection_session_id = worker.session_id if worker is not None else None
        self.selected_session_id = initial_session_id

        self.root = tk.Tk()
        self.root.title("Temperature Monitor")
        self.root.geometry("1180x760")
        self.root.protocol("WM_DELETE_WINDOW", self._on_close)

        self.session_var = tk.StringVar()
        self.sensor_var = tk.StringVar()
        self.status_var = tk.StringVar(value="Initializing...")
        self._session_labels: dict[str, int] = {}
        self._sensor_labels: dict[str, str] = {}

        self._build_ui()
        self._reload_sessions()
        self._reload_sensors()
        self._refresh_plot()
        self.root.after(250, self._tick)

    def run(self) -> None:
        self.root.mainloop()

    def _build_ui(self) -> None:
        controls = ttk.Frame(self.root, padding=12)
        controls.pack(fill=tk.X)

        ttk.Label(controls, text="Session").grid(row=0, column=0, sticky=tk.W, padx=(0, 8))
        self.session_combo = ttk.Combobox(
            controls,
            textvariable=self.session_var,
            state="readonly",
            width=48,
        )
        self.session_combo.grid(row=0, column=1, sticky=tk.EW, padx=(0, 16))
        self.session_combo.bind("<<ComboboxSelected>>", self._on_session_selected)

        ttk.Label(controls, text="Sensor").grid(row=0, column=2, sticky=tk.W, padx=(0, 8))
        self.sensor_combo = ttk.Combobox(
            controls,
            textvariable=self.sensor_var,
            state="readonly",
            width=48,
        )
        self.sensor_combo.grid(row=0, column=3, sticky=tk.EW, padx=(0, 16))
        self.sensor_combo.bind("<<ComboboxSelected>>", self._on_sensor_selected)

        refresh_button = ttk.Button(controls, text="Refresh", command=self._manual_refresh)
        refresh_button.grid(row=0, column=4, sticky=tk.E)

        controls.columnconfigure(1, weight=1)
        controls.columnconfigure(3, weight=1)

        status = ttk.Label(self.root, textvariable=self.status_var, padding=(12, 0, 12, 8))
        status.pack(fill=tk.X)

        self.figure = Figure(figsize=(11, 6), dpi=100)
        self.axes = self.figure.add_subplot(111)
        self.axes.set_title("Temperature over time")
        self.axes.set_xlabel("Time")
        self.axes.set_ylabel("Temperature (°C)")
        self.axes.grid(True, alpha=0.3)
        self.axes.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))

        canvas_frame = ttk.Frame(self.root, padding=(12, 0, 12, 12))
        canvas_frame.pack(fill=tk.BOTH, expand=True)

        self.canvas = FigureCanvasTkAgg(self.figure, master=canvas_frame)
        self.canvas.draw()
        self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True)

        toolbar = NavigationToolbar2Tk(self.canvas, canvas_frame, pack_toolbar=False)
        toolbar.update()
        toolbar.pack(fill=tk.X)

    def _tick(self) -> None:
        try:
            self._reload_sessions()
            if self.worker is not None:
                self.current_collection_session_id = self.worker.session_id
                self.status_var.set(self.worker.status_text())
            else:
                self.status_var.set("Viewing stored sessions. No active collector in this window.")
            if self.selected_session_id is None and self.current_collection_session_id is not None:
                self.selected_session_id = self.current_collection_session_id
                self._reload_sensors()
            if self.selected_session_id == self.current_collection_session_id:
                self._refresh_plot()
        except Exception as exc:  # pragma: no cover - GUI feedback
            self.status_var.set(f"Live mode error: {exc}")
        finally:
            self.root.after(500, self._tick)

    def _manual_refresh(self) -> None:
        self._reload_sessions()
        self._reload_sensors()
        self._refresh_plot()

    def _reload_sessions(self) -> None:
        sessions = self.store.list_sessions()
        labels = [format_session_label(session) for session in sessions]
        self._session_labels = {format_session_label(session): session.id for session in sessions}
        self.session_combo["values"] = labels

        selected_label = next(
            (
                label
                for label, session_id in self._session_labels.items()
                if session_id == self.selected_session_id
            ),
            labels[0] if labels else "",
        )
        self.session_var.set(selected_label)

    def _reload_sensors(self) -> None:
        if self.selected_session_id is None:
            self._sensor_labels = {}
            self.sensor_combo["values"] = []
            self.sensor_var.set("")
            return
        sensor_rows = self.store.get_sensor_rows(self.selected_session_id)
        labels = [f"{row['component']} / {row['name']}" for row in sensor_rows]
        self._sensor_labels = {
            f"{row['component']} / {row['name']}": str(row["sensor_key"]) for row in sensor_rows
        }
        self.sensor_combo["values"] = labels
        if not labels:
            self.sensor_var.set("")
            return
        if self.sensor_var.get() not in self._sensor_labels:
            self.sensor_var.set(labels[0])

    def _refresh_plot(self) -> None:
        if self.selected_session_id is None:
            self.axes.clear()
            self.axes.grid(True, alpha=0.3)
            self.axes.set_title("No session selected yet")
            self.canvas.draw_idle()
            return
        sensor_key = self._sensor_labels.get(self.sensor_var.get())
        self.axes.clear()
        self.axes.grid(True, alpha=0.3)
        self.axes.set_xlabel("Time")
        self.axes.set_ylabel("Temperature (°C)")
        self.axes.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))

        if sensor_key is None:
            self.axes.set_title("No sensor selected")
            self.canvas.draw_idle()
            return

        rows = self.store.get_measurements_for_sensor(self.selected_session_id, sensor_key)
        if not rows:
            self.axes.set_title("No measurements yet")
            self.canvas.draw_idle()
            return

        x_values = mdates.date2num([parse_timestamp(str(row["captured_at"])) for row in rows])
        y_values = [float(row["value"]) for row in rows]

        self.axes.plot(x_values, y_values, color="#d97706", linewidth=1.8)
        self.axes.fill_between(x_values, y_values, alpha=0.15, color="#f59e0b")
        self.axes.set_title(self.sensor_var.get())
        self.axes.relim()
        self.axes.autoscale_view()
        self.figure.autofmt_xdate()
        self.canvas.draw_idle()

    def _on_session_selected(self, _: object) -> None:
        session_id = self._session_labels.get(self.session_var.get())
        if session_id is None:
            return
        self.selected_session_id = session_id
        self._reload_sensors()
        self._refresh_plot()

    def _on_sensor_selected(self, _: object) -> None:
        self._refresh_plot()

    def _on_close(self) -> None:
        if self.worker is not None:
            try:
                self.worker.stop()
            except Exception as exc:  # pragma: no cover - GUI feedback
                messagebox.showerror("Temperature Monitor", str(exc))
        self.root.destroy()


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Cross-platform temperature monitoring with sessions, SQLite storage and live visualization."
    )
    parser.add_argument(
        "--db-path",
        type=Path,
        default=DEFAULT_DB_PATH,
        help=f"SQLite database path (default: {DEFAULT_DB_PATH})",
    )
    parser.add_argument(
        "--backend",
        choices=["auto", "psutil", "librehardwaremonitor"],
        default="auto",
        help="Metric backend to use.",
    )
    parser.add_argument(
        "--windows-lhm-dll",
        type=Path,
        help="Path to LibreHardwareMonitorLib.dll for Windows backend.",
    )
    parser.add_argument(
        "--no-admin-relaunch",
        action="store_true",
        help="Disable automatic UAC relaunch for Windows LibreHardwareMonitor collection modes.",
    )
    parser.add_argument(
        "--_elevated-relaunch",
        action="store_true",
        help=argparse.SUPPRESS,
    )

    subparsers = parser.add_subparsers(dest="command", required=True)

    watch_parser = subparsers.add_parser(
        "watch", help="Collect measurements in the foreground until stopped."
    )
    watch_parser.add_argument("-n", "--name", help="Optional human-readable session name.")
    watch_parser.add_argument(
        "-i",
        "--interval",
        type=float,
        default=5.0,
        help="Seconds between measurements.",
    )
    watch_parser.add_argument("--session", help="Resume an existing session by id or exact name.")

    live_parser = subparsers.add_parser(
        "live", help="Open a GUI for live plotting or browsing existing sessions."
    )
    live_parser.add_argument(
        "-n", "--name", help="Optional human-readable session name for a new live session."
    )
    live_parser.add_argument(
        "-i",
        "--interval",
        type=float,
        default=2.0,
        help="Seconds between live measurements when collecting.",
    )
    live_parser.add_argument("--session", help="Existing session id or exact name to open.")
    live_parser.add_argument(
        "--resume",
        action="store_true",
        help="When used with --session, continue collecting into that session.",
    )

    analyze_parser = subparsers.add_parser(
        "analyze", help="Open the GUI viewer without starting measurements."
    )
    analyze_parser.add_argument(
        "--session", help="Session id or exact name. Defaults to the latest session."
    )

    stats_parser = subparsers.add_parser("stats", help="Show basic statistics for each sensor.")
    stats_parser.add_argument(
        "--session", help="Session id or exact name. Defaults to the latest session."
    )

    export_parser = subparsers.add_parser("export", help="Export session data to CSV.")
    export_parser.add_argument("output", type=Path, help="CSV file path.")
    export_parser.add_argument(
        "--session", help="Session id or exact name. Defaults to the latest session."
    )

    return parser


def open_store(db_path: Path) -> MeasurementStore:
    db_path.parent.mkdir(parents=True, exist_ok=True)
    return MeasurementStore(db_path)


def run_watch(args: argparse.Namespace, store: MeasurementStore, factory: BackendFactory) -> int:
    requested_backend = args.backend
    resumed_session: SessionRecord | None = None
    if args.session:
        resumed_session = store.resolve_session(args.session)
        if requested_backend == "auto":
            requested_backend = resumed_session.backend_name
    backend = factory.create(requested_backend)
    try:
        if args.session:
            if resumed_session is None:
                raise RuntimeError("Failed to resolve the session to resume.")
            session = resumed_session
            store.clear_session_closed(session.id)
            recorder = Recorder(store, backend, session)
            sensors = recorder.prepare()
            print(
                f"Session {session.id} resumed with {len(sensors)} sensors. Press Ctrl+C to stop."
            )
        else:
            session, recorder, sensors, inserted = wait_for_bootstrap_session(
                store=store,
                backend=backend,
                session_name=args.name,
                metadata={"command": "watch", "interval_seconds": args.interval},
                interval_seconds=args.interval,
            )
            timestamp = utc_now_text()
            print(f"Session {session.id} ready with {len(sensors)} sensors. Press Ctrl+C to stop.")
            print(f"[{timestamp}] inserted {inserted} readings into session {session.id}")

        try:
            while True:
                time.sleep(args.interval)
                inserted = recorder.capture_once()
                timestamp = utc_now_text()
                print(f"[{timestamp}] inserted {inserted} readings into session {session.id}")
        except KeyboardInterrupt:
            print("\nStopping watch mode...")
            return 0
        finally:
            store.mark_session_closed(session.id)
    finally:
        backend.close()


def run_live(args: argparse.Namespace, store: MeasurementStore, factory: BackendFactory) -> int:
    worker: WatchWorker | None = None

    if args.session is None:
        initial_session_id = None
        worker = WatchWorker(
            db_path=store.db_path,
            factory=factory,
            backend_name=args.backend,
            session_id=None,
            interval_seconds=args.interval,
            session_name=args.name,
            metadata={"command": "live", "interval_seconds": args.interval},
        )
        worker.start()
    else:
        session = store.resolve_session(args.session)
        initial_session_id = session.id
        if args.resume:
            store.clear_session_closed(session.id)
            requested_backend = session.backend_name if args.backend == "auto" else args.backend
            worker = WatchWorker(
                db_path=store.db_path,
                factory=factory,
                backend_name=requested_backend,
                session_id=session.id,
                interval_seconds=args.interval,
            )
            worker.start()

    window = LiveMonitorWindow(
        store=store,
        initial_session_id=initial_session_id,
        worker=worker,
    )
    window.run()
    return 0


def run_analyze(args: argparse.Namespace, store: MeasurementStore) -> int:
    session = store.resolve_session(args.session)
    window = LiveMonitorWindow(
        store=store,
        initial_session_id=session.id,
        worker=None,
    )
    window.run()
    return 0


def run_stats(args: argparse.Namespace, store: MeasurementStore) -> int:
    session = store.resolve_session(args.session)
    rows = store.get_measurements_for_session(session.id)
    if not rows:
        raise RuntimeError(f"Session {session.id} has no measurements.")

    grouped: dict[str, dict[str, Any]] = {}
    for row in rows:
        sensor_key = str(row["sensor_key"])
        grouped.setdefault(
            sensor_key,
            {
                "name": str(row["name"]),
                "component": str(row["component"]),
                "unit": str(row["unit"]),
                "values": [],
            },
        )["values"].append(float(row["value"]))

    print(f"Session {session.id}: {session.name or '(unnamed)'}")
    for sensor in sorted(grouped.values(), key=lambda item: (item["component"], item["name"])):
        values = sensor["values"]
        q1, median, q3 = quantiles(values)
        stddev = statistics.pstdev(values) if len(values) > 1 else 0.0
        print()
        print(f"{sensor['component']} / {sensor['name']} [{sensor['unit']}]")
        print(f"  count: {len(values)}")
        print(f"  min: {min(values):.2f}")
        print(f"  max: {max(values):.2f}")
        print(f"  mean: {statistics.mean(values):.2f}")
        print(f"  std: {stddev:.2f}")
        print(f"  q1: {q1:.2f}")
        print(f"  median: {median:.2f}")
        print(f"  q3: {q3:.2f}")
    return 0


def run_export(args: argparse.Namespace, store: MeasurementStore) -> int:
    session = store.resolve_session(args.session)
    rows = store.get_measurements_for_session(session.id)
    if not rows:
        raise RuntimeError(f"Session {session.id} has no measurements.")

    args.output.parent.mkdir(parents=True, exist_ok=True)
    with args.output.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.writer(handle)
        writer.writerow(
            [
                "session_id",
                "session_name",
                "captured_at",
                "sensor_key",
                "sensor_name",
                "component",
                "unit",
                "value",
            ]
        )
        for row in rows:
            writer.writerow(
                [
                    session.id,
                    session.name or "",
                    row["captured_at"],
                    row["sensor_key"],
                    row["name"],
                    row["component"],
                    row["unit"],
                    row["value"],
                ]
            )
    print(f"Exported session {session.id} to {args.output}")
    return 0


def validate_args(args: argparse.Namespace) -> None:
    if getattr(args, "interval", 1.0) <= 0:
        raise RuntimeError("Interval must be greater than zero.")
    if args.command == "live" and args.resume and args.session is None:
        raise RuntimeError("--resume requires --session.")


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    validate_args(args)
    if should_auto_relaunch_as_admin(args):
        return relaunch_as_admin()

    store = open_store(args.db_path)
    try:
        factory = BackendFactory(windows_lhm_dll=args.windows_lhm_dll)
        if args.command == "watch":
            return run_watch(args, store, factory)
        if args.command == "live":
            return run_live(args, store, factory)
        if args.command == "analyze":
            return run_analyze(args, store)
        if args.command == "stats":
            return run_stats(args, store)
        if args.command == "export":
            return run_export(args, store)
        raise RuntimeError(f"Unsupported command: {args.command}")
    finally:
        store.close()


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as exc:
        if "--_elevated-relaunch" in sys.argv:
            show_windows_error_dialog("Temperature Monitor Error", str(exc))
        print(f"Error: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc