"""Normalized serial interface for fbuild and pyserial backends.

Provides a common async interface that both pyserial and fbuild serial monitors
can be accessed through. The fbuild adapter offloads sync calls to a thread
executor when running inside an async event loop.
"""

from __future__ import annotations

import _thread
import asyncio
import time
import warnings
from collections.abc import AsyncIterator
from concurrent.futures import ThreadPoolExecutor
from typing import Protocol, runtime_checkable

import serial as pyserial

from ci.util.global_interrupt_handler import handle_keyboard_interrupt


def _pyserial_dtr_reset(port: str) -> bool:
    """Reset an ESP32 device using the esptool ClassicReset DTR/RTS sequence.

    Sequence: DTR=false,RTS=true (hold EN low) → DTR=true,RTS=false (release)
    → DTR=false (final state). This matches esptool's --before default-reset.
    """
    try:
        s = pyserial.Serial(port, 115200, timeout=0.1)
        try:
            s.dtr = False
            s.rts = True
            time.sleep(0.1)
            s.dtr = True
            s.rts = False
            time.sleep(0.05)
            s.dtr = False
        finally:
            s.close()
        return True
    except (OSError, pyserial.SerialException):
        return False


@runtime_checkable
class SerialInterface(Protocol):
    """Async serial interface protocol.

    All serial communication in the RPC layer goes through this interface,
    normalizing access between pyserial and fbuild backends.
    """

    async def connect(self) -> None:
        """Open serial connection."""
        ...

    async def close(self) -> None:
        """Close serial connection."""
        ...

    async def write(self, data: str) -> None:
        """Write string data to serial port."""
        ...

    async def read_lines(self, timeout: float) -> AsyncIterator[str]:
        """Async iterator yielding lines from serial port.

        Args:
            timeout: Maximum time to wait for lines in seconds.

        Yields:
            Lines from serial output (without newlines).
        """
        ...
        # Make this a valid async generator for protocol purposes
        if False:  # pragma: no cover
            yield ""

    async def reset_device(self, board: str | None) -> bool:
        """Reset the device via DTR/RTS toggling.

        Args:
            board: Board identifier for platform-specific reset sequence.
                   Pass None for generic DTR toggle.

        Returns:
            True if reset succeeded, False otherwise.
        """
        ...


class PySerialAdapter:
    """Adapter wrapping pyserial_monitor.SerialMonitor as SerialInterface."""

    def __init__(
        self,
        port: str,
        baud_rate: int = 115200,
        auto_reconnect: bool = True,
        verbose: bool = False,
    ) -> None:
        from ci.util.pyserial_monitor import SerialMonitor

        self._monitor = SerialMonitor(
            port=port,
            baud_rate=baud_rate,
            auto_reconnect=auto_reconnect,
            verbose=verbose,
        )

    async def connect(self) -> None:
        self._monitor.__enter__()

    async def close(self) -> None:
        self._monitor.__exit__(None, None, None)

    async def write(self, data: str) -> None:
        self._monitor.write(data)

    async def read_lines(self, timeout: float) -> AsyncIterator[str]:
        for line in self._monitor.read_lines(timeout=timeout):
            yield line

    async def reset_device(self, board: str | None) -> bool:
        """Reset device via direct pyserial DTR/RTS toggle."""
        return _pyserial_dtr_reset(self._monitor.port)


class FbuildSerialAdapter:
    """Adapter wrapping fbuild SerialMonitor as SerialInterface.

    fbuild's SerialMonitor uses run_until_complete() internally, which crashes
    when called from an already-running async event loop. This adapter offloads
    all sync calls to a ThreadPoolExecutor to avoid the conflict.
    """

    _warned: bool = False

    def __init__(
        self,
        port: str,
        baud_rate: int = 115200,
        auto_reconnect: bool = True,
        verbose: bool = False,
    ) -> None:
        from fbuild.api import SerialMonitor

        self._monitor = SerialMonitor(
            port=port,
            baud_rate=baud_rate,
            auto_reconnect=auto_reconnect,
            verbose=verbose,
        )
        self._executor = ThreadPoolExecutor(max_workers=1)

    def _warn_once(self) -> None:
        if not FbuildSerialAdapter._warned:
            FbuildSerialAdapter._warned = True
            warnings.warn(
                "Converting sync serial monitor (fbuild) to async via thread "
                "executor. This will be removed when fbuild provides native "
                "async support.",
                stacklevel=3,
            )

    async def _run_in_thread(self, func, *args):  # type: ignore[no-untyped-def]
        self._warn_once()
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(self._executor, func, *args)

    async def connect(self) -> None:
        await self._run_in_thread(self._monitor.__enter__)

    async def close(self) -> None:
        await self._run_in_thread(self._monitor.__exit__, None, None, None)
        # NOTE: do NOT shut down self._executor here. The autoresearch
        # recovery path (ci/autoresearch/gpio.py) calls close() then reuses
        # the same adapter on a new RpcClient. Shutting the executor down
        # makes the next _run_in_thread fail with "cannot schedule new
        # futures after shutdown" and breaks DTR-reset retries. See
        # FastLED/fbuild#592.

    async def write(self, data: str) -> None:
        await self._run_in_thread(self._monitor.write, data)

    async def read_lines(self, timeout: float) -> AsyncIterator[str]:
        loop = asyncio.get_running_loop()
        queue: asyncio.Queue[str | None] = asyncio.Queue()

        def _producer() -> None:
            # fbuild's native read_lines() returns the first batch of lines
            # then exits (it does NOT keep reading until timeout). We must
            # loop over multiple read_lines() calls to cover the full timeout.
            deadline = time.monotonic() + timeout
            try:
                while True:
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        break
                    for line in self._monitor.read_lines(timeout=remaining):
                        loop.call_soon_threadsafe(queue.put_nowait, line)
            except KeyboardInterrupt:
                _thread.interrupt_main()
                raise
            except Exception as exc:
                warnings.warn(
                    f"FbuildSerialAdapter read_lines error: {exc}",
                    stacklevel=2,
                )
            finally:
                loop.call_soon_threadsafe(queue.put_nowait, None)

        # Start producer in thread
        self._executor.submit(_producer)

        while True:
            item = await queue.get()
            if item is None:
                break
            yield item

    async def reset_device(self, board: str | None) -> bool:
        """Reset device via fbuild daemon's reset endpoint.

        Uses wait_for_output=True to block until the device produces
        serial output (rebooted and running), avoiding fixed sleep delays.
        """
        if not hasattr(self._monitor, "reset_device"):
            return _pyserial_dtr_reset(self._monitor.port)
        try:
            result = await self._run_in_thread(
                self._monitor.reset_device, board, True, 5.0
            )
            return bool(result)
        except KeyboardInterrupt as ki:
            handle_keyboard_interrupt(ki)
            raise
        except Exception:
            return False


def create_serial_interface(
    port: str,
    baud_rate: int = 115200,
    use_pyserial: bool = False,
    auto_reconnect: bool = True,
    verbose: bool = False,
) -> SerialInterface:
    """Factory function to create the appropriate serial interface.

    Args:
        port: Serial port path (e.g., "/dev/ttyUSB0", "COM3")
        baud_rate: Serial baud rate (default: 115200)
        use_pyserial: If True, use pyserial directly instead of fbuild
        auto_reconnect: Enable auto-reconnect (passed to underlying monitor)
        verbose: Enable verbose debug output

    Returns:
        SerialInterface implementation (PySerialAdapter or FbuildSerialAdapter)
    """
    if use_pyserial:
        return PySerialAdapter(
            port=port,
            baud_rate=baud_rate,
            auto_reconnect=auto_reconnect,
            verbose=verbose,
        )
    else:
        return FbuildSerialAdapter(
            port=port,
            baud_rate=baud_rate,
            auto_reconnect=auto_reconnect,
            verbose=verbose,
        )
