escpos-vfd 0.2.0

//! Tokio API поверх настоящего `AsyncWrite`.
//!
//! Модуль доступен только с feature `tokio`. Он повторяет sync API, но использует
//! `tokio-serial`, `tokio::sync::mpsc`, `oneshot`-подтверждения и async sleep для
//! задержек яркости и бегущей строки. Низкоуровневый `AsyncVfd` работает с любым
//! `AsyncWrite`, поэтому тесты и нестандартные транспорты не требуют настоящего
//! serial-порта.

use crate::codec::{EpsonCodec, changed_runs, fit_to_width, replace_cached_range, sanitize_text};
use crate::config::{DisplaySettings, VfdConfig};
use crate::error::{ConfigError, Result, VfdError};
use ::tokio::io::{AsyncWrite, AsyncWriteExt};
use ::tokio::sync::{mpsc, oneshot};
use ::tokio::task::JoinHandle;
use ::tokio::time::{Duration, Instant, sleep, sleep_until};
use serialport::SerialPortBuilder;
use tokio_serial::{SerialPortBuilderExt, SerialStream};

type Ack = oneshot::Sender<Result<()>>;

enum Cmd {
    Clear {
        ack: Ack,
    },
    PrintLine {
        line: u8,
        text: String,
        ack: Ack,
    },
    PrintLineDiff {
        line: u8,
        text: String,
        ack: Ack,
    },
    PrintAt {
        x: u8,
        y: u8,
        text: String,
        ack: Ack,
    },
    WriteRaw {
        bytes: Vec<u8>,
        ack: Ack,
    },
    SetMarqueeText {
        text: String,
        ack: Ack,
    },
    StartMarquee {
        line: u8,
        cps: u32,
        end_pause: Duration,
        ack: Ack,
    },
    StopMarquee {
        ack: Ack,
    },
    SetBrightness {
        level: u8,
        ack: Ack,
    },
    Shutdown {
        ack: Ack,
    },
}

/// Async-драйвер VFD поверх `AsyncWrite`.
///
/// Драйвер выполняет запись напрямую в текущем async task и не использует
/// `spawn_blocking`. Для сериализации команд из разных задач используйте
/// [`AsyncVfdWorker`]. Все координаты задаются от единицы.
pub struct AsyncVfd<T: AsyncWrite + Unpin = SerialStream> {
    transport: T,
    codec: EpsonCodec,
}

impl AsyncVfd<SerialStream> {
    /// Открывает serial-порт через `tokio-serial` и инициализирует дисплей.
    ///
    /// Метод проверяет [`VfdConfig`], открывает serial-порт и отправляет init-команды
    /// (`ESC @`, `ESC t n`) согласно [`DisplaySettings`].
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::Config`] при неверной конфигурации,
    /// [`VfdError::Serial`] при ошибке открытия порта и [`VfdError::Io`] при ошибке
    /// async-записи init-команд.
    pub async fn open(cfg: VfdConfig) -> Result<Self> {
        cfg.validate()?;
        let serial = cfg.serial;
        let display = cfg.display;

        let mut builder: SerialPortBuilder = tokio_serial::new(&serial.port_name, serial.baud_rate)
            .data_bits(serial.data_bits)
            .parity(serial.parity)
            .stop_bits(serial.stop_bits)
            .flow_control(serial.flow_control)
            .timeout(serial.timeout);
        #[cfg(unix)]
        {
            builder = builder.exclusive(serial.exclusive);
        }

        let port = builder.open_native_async()?;
        Self::from_transport(port, display).await
    }
}

impl<T: AsyncWrite + Unpin> AsyncVfd<T> {
    /// Создаёт async-драйвер поверх произвольного async-транспорта.
    ///
    /// Как и sync-вариант, метод отправляет init-последовательность сразу после проверки
    /// [`DisplaySettings`].
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::Config`] для неверных настроек дисплея или
    /// [`VfdError::Io`] при ошибке async-записи.
    ///
    /// # Примеры
    ///
    /// ```
    /// # #[cfg(feature = "tokio")]
    /// # async fn demo() -> escpos_vfd::Result<()> {
    /// use escpos_vfd::{DisplaySettings, TextEncoding};
    /// use escpos_vfd::tokio::AsyncVfd;
    ///
    /// let display = DisplaySettings::new(4, 1, TextEncoding::Ascii);
    /// let mut vfd = AsyncVfd::from_transport(Vec::<u8>::new(), display).await?;
    /// vfd.print_line(1, "OK").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn from_transport(mut transport: T, display: DisplaySettings) -> Result<Self> {
        display.validate()?;
        let codec = EpsonCodec::new(display);
        let init = codec.init();
        if !init.is_empty() {
            transport.write_all(&init).await?;
        }
        Ok(Self { transport, codec })
    }

    /// Настройки дисплея.
    pub fn display(&self) -> &DisplaySettings {
        self.codec.display()
    }

    /// Количество символов в строке.
    pub fn columns(&self) -> usize {
        self.display().columns
    }

    /// Количество строк.
    pub fn rows(&self) -> usize {
        self.display().rows
    }

    /// Очищает дисплей.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::Io`] при ошибке async-записи.
    pub async fn clear(&mut self) -> Result<()> {
        self.transport.write_all(&self.codec.clear()).await?;
        Ok(())
    }

    /// Полностью перезаписывает строку.
    ///
    /// Текст санитизируется, обрезается и дополняется пробелами до ширины дисплея.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::InvalidLine`] или [`VfdError::Io`].
    pub async fn print_line(&mut self, line: u8, text: &str) -> Result<()> {
        self.codec.validate_line(line)?;
        self.goto_xy(1, line).await?;
        let fitted = self.codec.fit_line(text);
        self.write_text(&fitted).await
    }

    /// Выводит подготовленный кадр.
    ///
    /// Метод не выполняет санацию текста. Используйте его для уже подготовленных кадров.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::InvalidLine`] или [`VfdError::Io`].
    pub async fn print_frame(&mut self, line: u8, frame: &str) -> Result<()> {
        self.codec.validate_line(line)?;
        self.goto_xy(1, line).await?;
        let fitted = fit_to_width(frame, self.columns());
        self.write_text(&fitted).await
    }

    /// Печатает текст с координаты `(x, y)`.
    ///
    /// Координаты задаются от единицы. Текст санитизируется и обрезается по правому краю
    /// текущей строки.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::InvalidCoordinate`] или [`VfdError::Io`].
    pub async fn print_at(&mut self, x: u8, y: u8, text: &str) -> Result<()> {
        let text = sanitize_text(text);
        self.print_at_prepared(x, y, &text).await
    }

    async fn print_at_prepared(&mut self, x: u8, y: u8, text: &str) -> Result<()> {
        self.codec.validate_xy(x, y)?;
        let remaining = self.columns() - usize::from(x) + 1;
        let text: String = text.chars().take(remaining).collect();
        self.goto_xy(x, y).await?;
        self.write_text(&text).await
    }

    /// Записывает байты напрямую.
    ///
    /// Байты не кодируются и не интерпретируются библиотекой. Это escape hatch для
    /// нестандартных ESC/POS-команд конкретного дисплея.
    pub async fn write_raw(&mut self, bytes: &[u8]) -> Result<()> {
        self.transport.write_all(bytes).await?;
        Ok(())
    }

    /// Устанавливает яркость.
    ///
    /// После записи команды выполняется async `flush`, затем async sleep на
    /// `display.brightness_settle`, если задержка не нулевая.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::UnsupportedBrightness`] или [`VfdError::Io`].
    pub async fn set_brightness(&mut self, level: u8) -> Result<()> {
        let cmd = self.codec.brightness(level)?;
        self.transport.write_all(&cmd).await?;
        self.transport.flush().await?;
        let settle = self.display().brightness_settle;
        if !settle.is_zero() {
            sleep(settle).await;
        }
        Ok(())
    }

    /// Flush async-транспорта.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::Io`], если внутренний транспорт не смог выполнить flush.
    pub async fn flush(&mut self) -> Result<()> {
        self.transport.flush().await?;
        Ok(())
    }

    /// Возвращает внутренний транспорт.
    ///
    /// Метод потребляет драйвер и возвращает ownership транспорта вызывающему коду.
    pub fn into_inner(self) -> T {
        self.transport
    }

    async fn goto_xy(&mut self, x: u8, y: u8) -> Result<()> {
        let cmd = self.codec.goto_xy(x, y)?;
        self.transport.write_all(&cmd).await?;
        Ok(())
    }

    async fn write_text(&mut self, s: &str) -> Result<()> {
        let bytes = self.codec.encode_text(s);
        self.transport.write_all(&bytes).await?;
        Ok(())
    }
}

/// Async handle с bounded queue и подтверждением выполнения I/O.
///
/// После того как команда принята очередью, отмена ожидающего future не отменяет уже
/// поставленную запись в устройство. Ошибка I/O возвращается через `Result`.
/// Клоны handle можно передавать в другие async tasks; backpressure создаёт `.await`
/// на отправке, когда очередь заполнена.
#[derive(Clone)]
pub struct AsyncVfdHandle {
    tx: mpsc::Sender<Cmd>,
}

impl AsyncVfdHandle {
    /// Очищает дисплей.
    ///
    /// # Ошибки
    ///
    /// Возвращает ошибки очереди, остановленного worker-а или async I/O.
    pub async fn clear(&self) -> Result<()> {
        self.call(|ack| Cmd::Clear { ack }).await
    }

    /// Устанавливает яркость.
    ///
    /// Future завершается после записи, flush и задержки `brightness_settle`.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::UnsupportedBrightness`], ошибки очереди или I/O.
    pub async fn set_brightness(&self, level: u8) -> Result<()> {
        self.call(|ack| Cmd::SetBrightness { level, ack }).await
    }

    /// Полностью перезаписывает строку.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::InvalidLine`], ошибки очереди или I/O.
    pub async fn print_line(&self, line: u8, text: impl Into<String>) -> Result<()> {
        let text = text.into();
        self.call(|ack| Cmd::PrintLine { line, text, ack }).await
    }

    /// Обновляет только изменившиеся диапазоны строки.
    ///
    /// Worker хранит кэш строк и отправляет только изменившиеся смежные диапазоны. На
    /// строке с активной marquee команда подтверждается без записи.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::InvalidLine`], ошибки очереди или I/O.
    pub async fn print_line_diff(&self, line: u8, text: impl Into<String>) -> Result<()> {
        let text = text.into();
        self.call(|ack| Cmd::PrintLineDiff { line, text, ack })
            .await
    }

    /// Печатает текст с координаты `(x, y)`.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::InvalidCoordinate`], ошибки очереди или I/O.
    pub async fn print_at(&self, x: u8, y: u8, text: impl Into<String>) -> Result<()> {
        let text = text.into();
        self.call(|ack| Cmd::PrintAt { x, y, text, ack }).await
    }

    /// Записывает байты напрямую.
    ///
    /// Raw-байты не обновляют строковый кэш worker.
    ///
    /// # Ошибки
    ///
    /// Возвращает ошибки очереди или [`VfdError::Io`].
    pub async fn write_raw(&self, bytes: impl Into<Vec<u8>>) -> Result<()> {
        let bytes = bytes.into();
        self.call(|ack| Cmd::WriteRaw { bytes, ack }).await
    }

    /// Заменяет текст бегущей строки.
    ///
    /// Если marquee уже активна, поток символов перестраивается сразу. Сам метод не
    /// пишет кадр синхронно; запись произойдёт по таймеру worker-а.
    ///
    /// # Ошибки
    ///
    /// Возвращает ошибки очереди или остановленного worker-а.
    pub async fn set_marquee_text(&self, text: impl Into<String>) -> Result<()> {
        let text = text.into();
        self.call(|ack| Cmd::SetMarqueeText { text, ack }).await
    }

    /// Запускает бегущую строку.
    ///
    /// `cps` - скорость в символах в секунду; `0` приводится к `1`. `end_pause` задаёт
    /// async-паузу в конце полного прохода текста.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::InvalidLine`] или ошибки очереди.
    pub async fn start_marquee(&self, line: u8, cps: u32, end_pause: Duration) -> Result<()> {
        self.call(|ack| Cmd::StartMarquee {
            line,
            cps,
            end_pause,
            ack,
        })
        .await
    }

    /// Останавливает бегущую строку.
    ///
    /// Останавливает таймер marquee. Видимый текст на дисплее не очищается автоматически.
    ///
    /// # Ошибки
    ///
    /// Возвращает ошибки очереди или остановленного worker-а.
    pub async fn stop_marquee(&self) -> Result<()> {
        self.call(|ack| Cmd::StopMarquee { ack }).await
    }

    /// Завершает worker после ранее принятых команд.
    ///
    /// Чтобы дождаться завершения task и получить внутренний драйвер, используйте
    /// [`AsyncVfdWorker::shutdown`].
    ///
    /// # Ошибки
    ///
    /// Возвращает ошибки очереди или остановленного worker-а.
    pub async fn shutdown(&self) -> Result<()> {
        self.call(|ack| Cmd::Shutdown { ack }).await
    }

    async fn call(&self, build: impl FnOnce(Ack) -> Cmd) -> Result<()> {
        let (ack_tx, ack_rx) = oneshot::channel();
        self.tx
            .send(build(ack_tx))
            .await
            .map_err(|_| VfdError::QueueClosed)?;
        ack_rx.await.map_err(|_| VfdError::WorkerStopped)?
    }
}

/// Владеет async task записи в дисплей.
///
/// `AsyncVfdWorker` запускает одну задачу-писатель и возвращает клоны [`AsyncVfdHandle`]
/// для вызывающего кода. При `Drop` незавершённая задача отменяется без блокировки runtime.
pub struct AsyncVfdWorker<T: AsyncWrite + Unpin + Send + 'static = SerialStream> {
    handle: AsyncVfdHandle,
    join: Option<JoinHandle<Result<AsyncVfd<T>>>>,
}

impl AsyncVfdWorker<SerialStream> {
    /// Открывает serial-порт и запускает async worker.
    ///
    /// # Ошибки
    ///
    /// Возвращает ошибки конфигурации, открытия serial-порта или init-записи из
    /// [`AsyncVfd::open`].
    pub async fn start(cfg: VfdConfig) -> Result<Self> {
        cfg.validate()?;
        let capacity = cfg.queue_capacity;
        let vfd = AsyncVfd::open(cfg).await?;
        Self::from_vfd(vfd, capacity)
    }
}

impl<T: AsyncWrite + Unpin + Send + 'static> AsyncVfdWorker<T> {
    /// Запускает worker поверх готового async-драйвера.
    ///
    /// Worker сначала очищает дисплей в своей task, затем обрабатывает очередь команд.
    ///
    /// # Ошибки
    ///
    /// Возвращает [`VfdError::Config`], если `queue_capacity = 0`.
    pub fn from_vfd(vfd: AsyncVfd<T>, queue_capacity: usize) -> Result<Self> {
        if queue_capacity == 0 {
            return Err(ConfigError::ZeroQueueCapacity.into());
        }
        let (tx, rx) = mpsc::channel(queue_capacity);
        let handle = AsyncVfdHandle { tx };
        let join = ::tokio::spawn(async move { writer_loop(vfd, rx).await });

        Ok(Self {
            handle,
            join: Some(join),
        })
    }

    /// Асинхронно создаёт worker поверх произвольного транспорта.
    ///
    /// Init-команды записываются до запуска worker task.
    ///
    /// # Ошибки
    ///
    /// Возвращает ошибки [`AsyncVfd::from_transport`] или
    /// [`VfdError::Config`] при `queue_capacity = 0`.
    pub async fn from_transport(
        transport: T,
        display: DisplaySettings,
        queue_capacity: usize,
    ) -> Result<Self> {
        if queue_capacity == 0 {
            return Err(ConfigError::ZeroQueueCapacity.into());
        }
        let vfd = AsyncVfd::from_transport(transport, display).await?;
        Self::from_vfd(vfd, queue_capacity)
    }

    /// Возвращает handle.
    ///
    /// Клон можно передать в другие async tasks. Сам [`AsyncVfdWorker`] должен жить до
    /// явного [`AsyncVfdWorker::shutdown`], иначе `Drop` отменит task.
    pub fn handle(&self) -> AsyncVfdHandle {
        self.handle.clone()
    }

    /// Graceful shutdown с ожиданием task.
    ///
    /// Метод отправляет shutdown-команду, ждёт её подтверждения и затем ожидает join handle,
    /// возвращая внутренний [`AsyncVfd`] вместе с транспортом.
    ///
    /// # Ошибки
    ///
    /// Возвращает ошибку отправки shutdown, последнюю ошибку worker-а или
    /// [`VfdError::WorkerCancelled`], если task была отменена до завершения.
    pub async fn shutdown(mut self) -> Result<AsyncVfd<T>> {
        let shutdown_result = self.handle.shutdown().await;
        let worker_result = self
            .join
            .take()
            .expect("join handle exists")
            .await
            .map_err(|_| VfdError::WorkerCancelled)?;

        let vfd = worker_result?;
        shutdown_result?;
        Ok(vfd)
    }
}

impl<T: AsyncWrite + Unpin + Send + 'static> Drop for AsyncVfdWorker<T> {
    fn drop(&mut self) {
        if let Some(join) = self.join.take() {
            join.abort();
        }
    }
}

#[derive(Debug, Clone)]
struct MarqueeState {
    active: bool,
    line: u8,
    cps: u32,
    end_pause: Duration,
    text: String,
    stream: Vec<char>,
    offset: usize,
    paused_until: Option<Instant>,
    next_step: Option<Instant>,
}

impl MarqueeState {
    fn new() -> Self {
        Self {
            active: false,
            line: 1,
            cps: 5,
            end_pause: Duration::from_millis(1500),
            text: String::new(),
            stream: Vec::new(),
            offset: 0,
            paused_until: None,
            next_step: None,
        }
    }

    fn rebuild_stream(&mut self, width: usize) {
        let text = sanitize_text(&self.text);
        self.stream.clear();
        self.stream.reserve(width * 2 + text.chars().count());
        self.stream.extend(std::iter::repeat_n(' ', width));
        self.stream.extend(text.chars());
        self.stream.extend(std::iter::repeat_n(' ', width));
        self.offset = 0;
        self.paused_until = None;
        self.next_step = Some(Instant::now() + self.step_interval());
    }

    fn step_interval(&self) -> Duration {
        let cps = u64::from(self.cps.max(1));
        Duration::from_nanos((1_000_000_000 / cps).max(1))
    }

    fn next_deadline(&self) -> Option<Instant> {
        if !self.active {
            return None;
        }
        self.paused_until.or(self.next_step)
    }
}

async fn writer_loop<T: AsyncWrite + Unpin>(
    mut vfd: AsyncVfd<T>,
    mut rx: mpsc::Receiver<Cmd>,
) -> Result<AsyncVfd<T>> {
    vfd.clear().await?;
    let rows = vfd.rows();
    let mut marquee = MarqueeState::new();
    let mut last_lines = vec![String::new(); rows];

    loop {
        let event = match marquee.next_deadline() {
            Some(deadline) => {
                ::tokio::select! {
                    cmd = rx.recv() => match cmd {
                        Some(cmd) => WorkerEvent::Command(cmd),
                        None => WorkerEvent::Closed,
                    },
                    _ = sleep_until(deadline) => WorkerEvent::Timer,
                }
            }
            None => match rx.recv().await {
                Some(cmd) => WorkerEvent::Command(cmd),
                None => WorkerEvent::Closed,
            },
        };

        match event {
            WorkerEvent::Command(cmd) => {
                if handle_command(cmd, &mut vfd, &mut marquee, &mut last_lines).await? {
                    break;
                }
            }
            WorkerEvent::Timer => {
                render_marquee(&mut vfd, &mut marquee, &mut last_lines).await?;
            }
            WorkerEvent::Closed => break,
        }
    }

    Ok(vfd)
}

enum WorkerEvent {
    Command(Cmd),
    Timer,
    Closed,
}

async fn handle_command<T: AsyncWrite + Unpin>(
    cmd: Cmd,
    vfd: &mut AsyncVfd<T>,
    marquee: &mut MarqueeState,
    last_lines: &mut [String],
) -> Result<bool> {
    let width = vfd.columns();
    let rows = vfd.rows();
    match cmd {
        Cmd::Clear { ack } => {
            let result = vfd.clear().await;
            if result.is_ok() {
                last_lines.fill(String::new());
            }
            send_ack(ack, result);
        }
        Cmd::SetBrightness { level, ack } => send_ack(ack, vfd.set_brightness(level).await),
        Cmd::PrintLine { line, text, ack } => {
            let result = vfd.print_line(line, &text).await;
            if result.is_ok() {
                last_lines[(line - 1) as usize] = fit_to_width(&sanitize_text(&text), width);
            }
            send_ack(ack, result);
        }
        Cmd::PrintLineDiff { line, text, ack } => {
            let result = print_line_diff(vfd, marquee, last_lines, line, &text).await;
            send_ack(ack, result);
        }
        Cmd::PrintAt { x, y, text, ack } => {
            let result = print_at_cached(vfd, marquee, last_lines, x, y, &text).await;
            send_ack(ack, result);
        }
        Cmd::WriteRaw { bytes, ack } => send_ack(ack, vfd.write_raw(&bytes).await),
        Cmd::SetMarqueeText { text, ack } => {
            marquee.text = text;
            if marquee.active {
                marquee.rebuild_stream(width);
            }
            send_ack(ack, Ok(()));
        }
        Cmd::StartMarquee {
            line,
            cps,
            end_pause,
            ack,
        } => {
            let result = if line == 0 || usize::from(line) > rows {
                Err(VfdError::InvalidLine { line, rows })
            } else {
                last_lines[(line - 1) as usize].clear();
                marquee.active = true;
                marquee.line = line;
                marquee.cps = cps.max(1);
                marquee.end_pause = end_pause;
                marquee.rebuild_stream(width);
                Ok(())
            };
            send_ack(ack, result);
        }
        Cmd::StopMarquee { ack } => {
            if marquee.active && usize::from(marquee.line) <= last_lines.len() {
                last_lines[(marquee.line - 1) as usize].clear();
            }
            marquee.active = false;
            marquee.paused_until = None;
            marquee.next_step = None;
            send_ack(ack, Ok(()));
        }
        Cmd::Shutdown { ack } => {
            send_ack(ack, Ok(()));
            return Ok(true);
        }
    }
    Ok(false)
}

async fn print_line_diff<T: AsyncWrite + Unpin>(
    vfd: &mut AsyncVfd<T>,
    marquee: &MarqueeState,
    last_lines: &mut [String],
    line: u8,
    text: &str,
) -> Result<()> {
    if line == 0 || usize::from(line) > vfd.rows() {
        return Err(VfdError::InvalidLine {
            line,
            rows: vfd.rows(),
        });
    }
    if marquee.active && marquee.line == line {
        return Ok(());
    }

    let next = fit_to_width(&sanitize_text(text), vfd.columns());
    let idx = (line - 1) as usize;
    if last_lines[idx] == next {
        return Ok(());
    }
    if last_lines[idx].is_empty() {
        vfd.print_line(line, &next).await?;
        last_lines[idx] = next;
        return Ok(());
    }
    for (x, text) in changed_runs(&last_lines[idx], &next) {
        vfd.print_at_prepared(x, line, &text).await?;
    }
    last_lines[idx] = next;
    Ok(())
}

async fn print_at_cached<T: AsyncWrite + Unpin>(
    vfd: &mut AsyncVfd<T>,
    marquee: &MarqueeState,
    last_lines: &mut [String],
    x: u8,
    y: u8,
    text: &str,
) -> Result<()> {
    if x == 0 || y == 0 || usize::from(x) > vfd.columns() || usize::from(y) > vfd.rows() {
        return Err(VfdError::InvalidCoordinate {
            x,
            y,
            columns: vfd.columns(),
            rows: vfd.rows(),
        });
    }
    if marquee.active && marquee.line == y {
        return Ok(());
    }
    let remaining = vfd.columns() - usize::from(x) + 1;
    let text: String = sanitize_text(text).chars().take(remaining).collect();
    vfd.print_at_prepared(x, y, &text).await?;
    replace_cached_range(&mut last_lines[(y - 1) as usize], x, &text, vfd.columns());
    Ok(())
}

async fn render_marquee<T: AsyncWrite + Unpin>(
    vfd: &mut AsyncVfd<T>,
    marquee: &mut MarqueeState,
    last_lines: &mut [String],
) -> Result<()> {
    if !marquee.active {
        return Ok(());
    }
    let now = Instant::now();
    if let Some(until) = marquee.paused_until {
        if now < until {
            return Ok(());
        }
        marquee.paused_until = None;
        marquee.next_step = Some(now + marquee.step_interval());
        return Ok(());
    }
    if marquee.next_step.is_some_and(|deadline| now < deadline) {
        return Ok(());
    }

    let width = vfd.columns();
    if marquee.stream.len() < width {
        marquee.rebuild_stream(width);
    }
    let max_off = marquee.stream.len().saturating_sub(width);
    let start = marquee.offset.min(max_off);
    let end = (start + width).min(marquee.stream.len());
    let frame: String = marquee.stream[start..end].iter().collect();
    vfd.print_at_prepared(1, marquee.line, &frame).await?;
    if usize::from(marquee.line) <= last_lines.len() {
        last_lines[(marquee.line - 1) as usize] = frame;
    }

    if marquee.offset >= max_off {
        marquee.offset = 0;
        marquee.paused_until = Some(now + marquee.end_pause);
        marquee.next_step = None;
    } else {
        marquee.offset += 1;
        marquee.next_step = Some(now + marquee.step_interval());
    }
    Ok(())
}

fn send_ack(ack: Ack, result: Result<()>) {
    let _ = ack.send(result);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::{DisplaySettings, TextEncoding};
    use std::io;
    use std::pin::Pin;
    use std::sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    };
    use std::task::{Context, Poll};

    struct AsyncFailsAfterWrites {
        writes_left: usize,
        failed: Arc<AtomicBool>,
    }

    impl AsyncWrite for AsyncFailsAfterWrites {
        fn poll_write(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<io::Result<usize>> {
            if self.writes_left == 0 {
                self.failed.store(true, Ordering::SeqCst);
                return Poll::Ready(Err(io::Error::other("forced write failure")));
            }
            self.writes_left -= 1;
            Poll::Ready(Ok(buf.len()))
        }

        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
            Poll::Ready(Ok(()))
        }

        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
            Poll::Ready(Ok(()))
        }
    }

    #[tokio::test]
    async fn async_driver_writes_same_core_bytes() {
        let display = DisplaySettings::new(6, 2, TextEncoding::Ascii);
        let mut vfd = AsyncVfd::from_transport(Vec::<u8>::new(), display)
            .await
            .unwrap();

        vfd.print_line(2, "abc").await.unwrap();
        vfd.write_raw(&[0xAA]).await.unwrap();

        assert_eq!(
            vfd.into_inner(),
            vec![
                0x1B, 0x40, 0x1F, 0x24, 1, 2, b'a', b'b', b'c', b' ', b' ', b' ', 0xAA
            ]
        );
    }

    #[tokio::test(start_paused = true)]
    async fn async_worker_ack_and_marquee_scheduler() {
        let display = DisplaySettings::new(5, 2, TextEncoding::Ascii);
        let worker = AsyncVfdWorker::from_transport(Vec::<u8>::new(), display, 2)
            .await
            .unwrap();
        let handle = worker.handle();

        handle.set_marquee_text("abc").await.unwrap();
        handle
            .start_marquee(2, 10, Duration::from_millis(100))
            .await
            .unwrap();
        ::tokio::time::advance(Duration::from_millis(100)).await;
        ::tokio::task::yield_now().await;
        handle.stop_marquee().await.unwrap();

        let vfd = worker.shutdown().await.unwrap();
        assert!(
            vfd.into_inner()
                .windows(4)
                .any(|window| window == [0x1F, 0x24, 1, 2])
        );
    }

    #[tokio::test]
    async fn async_worker_rejects_zero_queue_capacity() {
        let display = DisplaySettings::new(5, 2, TextEncoding::Ascii);

        assert!(matches!(
            AsyncVfdWorker::from_transport(Vec::<u8>::new(), display, 0).await,
            Err(VfdError::Config(ConfigError::ZeroQueueCapacity))
        ));
    }

    #[tokio::test]
    async fn async_worker_print_at_validates_coordinates_before_marquee_skip() {
        let display = DisplaySettings::new(5, 2, TextEncoding::Ascii);
        let worker = AsyncVfdWorker::from_transport(Vec::<u8>::new(), display, 2)
            .await
            .unwrap();
        let handle = worker.handle();

        handle
            .start_marquee(2, 10, Duration::from_millis(100))
            .await
            .unwrap();

        assert!(matches!(
            handle.print_at(0, 2, "bad").await,
            Err(VfdError::InvalidCoordinate { x: 0, y: 2, .. })
        ));
        assert!(matches!(
            handle.print_at(6, 2, "bad").await,
            Err(VfdError::InvalidCoordinate { x: 6, y: 2, .. })
        ));
        handle.print_at(1, 2, "skipped").await.unwrap();

        worker.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn async_worker_shutdown_prefers_startup_io_error_over_closed_queue() {
        let display = DisplaySettings::new(5, 2, TextEncoding::Ascii);
        let failed = Arc::new(AtomicBool::new(false));
        let transport = AsyncFailsAfterWrites {
            writes_left: 1,
            failed: Arc::clone(&failed),
        };
        let worker = AsyncVfdWorker::from_transport(transport, display, 2)
            .await
            .unwrap();

        while !failed.load(Ordering::SeqCst) {
            ::tokio::task::yield_now().await;
        }

        assert!(matches!(worker.shutdown().await, Err(VfdError::Io(_))));
    }

    #[tokio::test(start_paused = true)]
    async fn async_worker_exits_when_channel_closes_with_active_marquee() {
        let display = DisplaySettings::new(5, 2, TextEncoding::Ascii);
        let worker = AsyncVfdWorker::from_transport(Vec::<u8>::new(), display, 2)
            .await
            .unwrap();
        let handle = worker.handle();

        handle.set_marquee_text("abc").await.unwrap();
        handle
            .start_marquee(2, 10, Duration::from_millis(100))
            .await
            .unwrap();
        drop(handle);
        let mut worker = worker;
        let join = worker.join.take().expect("join handle exists");
        drop(worker);

        let vfd = ::tokio::time::timeout(Duration::from_millis(1), async {
            join.await.map_err(|_| VfdError::WorkerCancelled)?
        })
        .await
        .expect("closed worker channel should terminate task")
        .unwrap();

        assert_eq!(vfd.rows(), 2);
    }
}