import logging
from pathlib import Path
from queue import Empty
import time

import pytest

import config
from control_runtime import (
    ControlWorker,
    build_open_loop_points,
    load_model,
    validate_device_settings,
    validate_opening,
    validate_open_loop_settings,
    validate_target,
)


class FakeHardware:
    def __init__(self, *, connect_ok=True):
        self.connect_ok = connect_ok
        self.connected = False
        self.disconnect_count = 0
        self.motor_commands = []

    def connect(self):
        self.connected = self.connect_ok
        return self.connect_ok

    def disconnect(self):
        self.connected = False
        self.disconnect_count += 1

    def get_flow(self, _address):
        return 20.0

    def get_pressure(self, address):
        if address == config.PRESSURE_BEFORE_ADDR:
            return 200.0
        return 100.0

    def set_motor_position(self, position, channel=None):
        self.motor_commands.append((float(position), channel))
        return True


class OverpressureHardware(FakeHardware):
    def get_pressure(self, address):
        if address == config.PRESSURE_BEFORE_ADDR:
            return config.MAX_PRESSURE_KPA + 1.0
        return 100.0


class MemoryOpenLoopRecorder:
    def __init__(self):
        self.csv_path = Path("memory_open_loop.csv")
        self.image_path = Path("memory_open_loop.png")
        self.sample_count = 0
        self.rows = []
        self.finalize_count = 0

    def record(self, **row):
        self.rows.append(row)
        self.sample_count += 1

    def finalize(self):
        self.finalize_count += 1
        return self.image_path if self.sample_count else None

    def close(self):
        pass


def wait_for_event(worker, predicate, timeout=2.0):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            event = worker.events.get(timeout=0.1)
        except Empty:
            continue
        if predicate(event):
            return event
    raise AssertionError("等待后台线程事件超时")


def make_logger():
    logger = logging.getLogger(f"test-control-runtime-{time.monotonic_ns()}")
    logger.addHandler(logging.NullHandler())
    return logger


def test_validate_ui_values():
    settings = validate_device_settings(" 192.168.1.88 ", "502", "1")
    assert settings.host == "192.168.1.88"
    assert settings.port == 502
    assert settings.slave_id == 1
    assert validate_target("50.5") == 50.5
    assert validate_opening("25.5") == 25.5
    with pytest.raises(ValueError, match="0～100"):
        validate_opening("100.1")


def test_builtin_model_is_valve_model_004():
    info = load_model()
    assert info.is_builtin is True
    assert info.path.name == "valve_model_004.json"
    assert info.point_count == 80
    assert info.model.motor_open == config.MOTOR_OPEN_POSITION
    assert info.model.motor_closed == config.MOTOR_CLOSED_POSITION


def test_validate_and_build_gui_open_loop_points():
    settings = validate_open_loop_settings(
        min_opening_pct="0",
        max_opening_pct="10",
        opening_step_pct="5",
        random_seed="42",
        include_fine_scan=False,
    )
    points = build_open_loop_points(settings)
    assert sorted(opening for opening, _position in points) == [0.0, 5.0, 10.0]
    assert all(
        config.MOTOR_OPEN_POSITION <= position <= config.MOTOR_CLOSED_POSITION
        for _opening, position in points
    )

    with pytest.raises(ValueError, match="开度范围"):
        validate_open_loop_settings(
            min_opening_pct=50,
            max_opening_pct=50,
            include_fine_scan=False,
        )


def test_worker_connect_control_and_safe_disconnect():
    hardware = FakeHardware()
    worker = ControlWorker(make_logger(), hardware_factory=lambda _settings: hardware)
    worker.start()
    wait_for_event(
        worker,
        lambda event: event.get("kind") == "state"
        and event.get("state") == "DISCONNECTED",
    )

    settings = validate_device_settings("192.168.1.88", 502, 1)
    worker.request_connect(settings)
    wait_for_event(
        worker,
        lambda event: event.get("kind") == "state"
        and event.get("state") == "MONITORING",
    )
    assert hardware.connected is True
    assert hardware.motor_commands[0] == (
        config.MOTOR_OPEN_POSITION,
        config.MOTOR_OUTPUT_ADDR,
    )

    worker.request_target(50.0)
    wait_for_event(
        worker,
        lambda event: event.get("kind") == "target_applied",
    )
    telemetry = wait_for_event(
        worker,
        lambda event: event.get("kind") == "telemetry"
        and event.get("target_flow_slm") == 50.0,
    )
    assert telemetry["measured_flow_slm"] == 20.0
    assert 0.0 <= telemetry["opening_pct"] <= 100.0
    assert telemetry["pid_mode"] in {"FAR", "NEAR"}

    worker.request_disconnect()
    wait_for_event(
        worker,
        lambda event: event.get("kind") == "state"
        and event.get("state") == "DISCONNECTED",
    )
    assert hardware.connected is False
    assert hardware.motor_commands[-1] == (
        config.MOTOR_OPEN_POSITION,
        config.MOTOR_OUTPUT_ADDR,
    )
    assert hardware.disconnect_count == 1

    worker.request_shutdown()
    wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
    worker.thread.join(timeout=1.0)
    assert not worker.is_alive()


def test_worker_runs_open_loop_scan_and_keeps_connection():
    hardware = FakeHardware()
    recorder = MemoryOpenLoopRecorder()
    worker = ControlWorker(
        make_logger(),
        hardware_factory=lambda _settings: hardware,
        open_loop_recorder_factory=lambda: recorder,
    )
    worker.start()
    wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
    worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
    wait_for_event(worker, lambda event: event.get("state") == "MONITORING")

    settings = validate_open_loop_settings(
        min_opening_pct=0,
        max_opening_pct=1,
        opening_step_pct=1,
        random_seed=1,
        sample_period_s=0.005,
        steady_window_s=0.01,
        max_wait_s=0.03,
        include_fine_scan=False,
    )
    worker.request_open_loop_start(settings)
    started = wait_for_event(
        worker, lambda event: event.get("kind") == "open_loop_started"
    )
    assert started["total_steps"] == 2
    finished = wait_for_event(
        worker, lambda event: event.get("kind") == "open_loop_finished"
    )
    assert finished["completed"] is True
    assert finished["sample_count"] == recorder.sample_count
    assert recorder.sample_count >= 4
    assert recorder.finalize_count == 1
    wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
    assert hardware.connected is True
    assert hardware.motor_commands[-1] == (
        config.MOTOR_OPEN_POSITION,
        config.MOTOR_OUTPUT_ADDR,
    )

    worker.request_shutdown()
    wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
    worker.thread.join(timeout=1.0)
    assert not worker.is_alive()


def test_worker_stops_closed_loop_at_full_open_and_accepts_manual_opening():
    hardware = FakeHardware()
    worker = ControlWorker(make_logger(), hardware_factory=lambda _settings: hardware)
    worker.start()
    wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
    worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
    wait_for_event(worker, lambda event: event.get("state") == "MONITORING")

    worker.request_target(50.0)
    wait_for_event(worker, lambda event: event.get("kind") == "target_applied")
    worker.request_manual_opening(25.0)
    warning = wait_for_event(
        worker,
        lambda event: event.get("kind") == "warning"
        and "先停止流量控制" in event.get("message", ""),
    )
    assert "先停止流量控制" in warning["message"]

    worker.request_stop_control()
    stopped = wait_for_event(
        worker, lambda event: event.get("kind") == "control_stopped"
    )
    assert stopped["opening_pct"] == 100.0
    assert stopped["motor_position"] == config.MOTOR_OPEN_POSITION
    wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
    assert hardware.motor_commands[-1] == (
        config.MOTOR_OPEN_POSITION,
        config.MOTOR_OUTPUT_ADDR,
    )

    worker.request_manual_opening(25.0)
    manual = wait_for_event(
        worker, lambda event: event.get("kind") == "manual_opening_applied"
    )
    assert manual["opening_pct"] == 25.0
    assert manual["motor_position"] == pytest.approx(950.0)
    wait_for_event(
        worker,
        lambda event: event.get("kind") == "telemetry"
        and event.get("pid_mode") == "未启用",
    )
    assert hardware.motor_commands[-1] == (950.0, config.MOTOR_OUTPUT_ADDR)

    worker.request_shutdown()
    wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
    worker.thread.join(timeout=1.0)
    assert not worker.is_alive()


def test_disconnect_during_open_loop_finalizes_partial_data():
    hardware = FakeHardware()
    recorder = MemoryOpenLoopRecorder()
    worker = ControlWorker(
        make_logger(),
        hardware_factory=lambda _settings: hardware,
        open_loop_recorder_factory=lambda: recorder,
    )
    worker.start()
    wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
    worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
    wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
    settings = validate_open_loop_settings(
        min_opening_pct=0,
        max_opening_pct=100,
        opening_step_pct=1,
        sample_period_s=0.02,
        steady_window_s=1,
        max_wait_s=1,
        include_fine_scan=False,
    )
    worker.request_open_loop_start(settings)
    wait_for_event(worker, lambda event: event.get("kind") == "open_loop_started")
    worker.request_disconnect()
    finished = wait_for_event(
        worker, lambda event: event.get("kind") == "open_loop_finished"
    )
    assert finished["completed"] is False
    assert finished["reason"] == "设备断开或程序退出"
    assert recorder.finalize_count == 1
    wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
    assert hardware.connected is False

    worker.request_shutdown()
    wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
    worker.thread.join(timeout=1.0)
    assert not worker.is_alive()


def test_worker_connection_failure_reports_fault_without_hardware_access():
    hardware = FakeHardware(connect_ok=False)
    worker = ControlWorker(make_logger(), hardware_factory=lambda _settings: hardware)
    worker.start()
    wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
    worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
    fault = wait_for_event(worker, lambda event: event.get("kind") == "fault")
    assert "无法连接 PLC" in fault["message"]
    assert hardware.motor_commands == []
    assert hardware.disconnect_count == 1

    worker.request_shutdown()
    wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
    worker.thread.join(timeout=1.0)
    assert not worker.is_alive()


def test_monitoring_stage_overpressure_triggers_safe_disconnect():
    hardware = OverpressureHardware()
    worker = ControlWorker(make_logger(), hardware_factory=lambda _settings: hardware)
    worker.start()
    wait_for_event(worker, lambda event: event.get("state") == "DISCONNECTED")
    worker.request_connect(validate_device_settings("192.168.1.88", 502, 1))
    wait_for_event(worker, lambda event: event.get("state") == "MONITORING")
    fault = wait_for_event(worker, lambda event: event.get("kind") == "fault")
    assert "超过上限" in fault["message"]
    assert hardware.connected is False
    assert hardware.motor_commands[-1][0] == config.MOTOR_OPEN_POSITION

    worker.request_shutdown()
    wait_for_event(worker, lambda event: event.get("kind") == "shutdown_complete")
    worker.thread.join(timeout=1.0)
    assert not worker.is_alive()