powerliners 0.1.1

// vim:fileencoding=utf-8:noet
//! Port of `powerline/lib/threaded.py`.
//!
//! Background-thread segment base classes. The simplest is
//! `MultiRunnedThread` (start/stop/join wrapper around a Python
//! `Thread`), used by VCS/network/weather segments that need to
//! refresh data off the render thread.
//!
//! This first chunk ports `MultiRunnedThread` faithfully. The richer
//! `ThreadedSegment` and `KwThreadedSegment` classes (which extend it
//! with periodic update + crash handling) land alongside the
//! `Segment` trait + Powerline logger — both are dispatch-heavy and
//! depend on substrate that isn't ported yet.

// from __future__ import (unicode_literals, division, absolute_import, print_function)  // py:2
// from threading import Thread, Lock, Event                                                // py:4
// from types import MethodType                                                              // py:5
// from powerline.lib.monotonic import monotonic                                              // py:7
// from powerline.segments import Segment                                                    // py:8

use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

/// Port of `class MultiRunnedThread` from
/// `powerline/lib/threaded.py:11`.
///
/// Python:
/// ```python
/// class MultiRunnedThread(object):
///     daemon = True
///
///     def __init__(self):
///         self.thread = None
///
///     def is_alive(self):
///         return self.thread and self.thread.is_alive()
///
///     def start(self):
///         self.shutdown_event.clear()
///         self.thread = Thread(target=self.run)
///         self.thread.daemon = self.daemon
///         self.thread.start()
///
///     def join(self, *args, **kwargs):
///         if self.thread:
///             return self.thread.join(*args, **kwargs)
///         return None
/// ```
///
/// Rust port: holds a `JoinHandle` plus a `shutdown_event` modeled as
/// `Arc<Mutex<bool>>`. The `run` method is provided by subclasses;
/// here we expose a `start_with` that accepts the run closure since
/// Rust traits can't model Python's `self.run()` indirection without
/// extra plumbing.
pub struct MultiRunnedThread {
    /// Python: `self.thread` — py:14
    pub thread: Mutex<Option<JoinHandle<()>>>,
    /// Python: `self.shutdown_event` — accessed by `start()` at py:20
    /// but populated by the subclass (typically `ThreadedSegment`). The
    /// Rust port carries it directly here to avoid the upstream
    /// "AttributeError until shutdown_event is set" footgun.
    pub shutdown_event: Arc<Mutex<bool>>,
    /// Python: class attribute `daemon = True` — py:12
    pub daemon: bool,
}

impl Default for MultiRunnedThread {
    fn default() -> Self {
        Self::new()
    }
}

impl MultiRunnedThread {
    /// Port of `MultiRunnedThread.__init__()` from
    /// `powerline/lib/threaded.py:14`.
    pub fn new() -> Self {
        // py:11  class MultiRunnedThread(object):
        // py:12  daemon = True
        // py:14  def __init__(self):
        // py:15  self.thread = None
        Self {
            thread: Mutex::new(None),
            shutdown_event: Arc::new(Mutex::new(false)),
            daemon: true,
        }
    }

    /// Port of `MultiRunnedThread.is_alive()` from
    /// `powerline/lib/threaded.py:16`.
    pub fn is_alive(&self) -> bool {
        // py:17  def is_alive(self):
        // py:18  return self.thread and self.thread.is_alive()
        let t = self.thread.lock().unwrap();
        match t.as_ref() {
            None => false,
            Some(h) => !h.is_finished(),
        }
    }

    /// Port of `MultiRunnedThread.start()` from
    /// `powerline/lib/threaded.py:19-24`.
    ///
    /// Python's `start()` dispatches via `self.run` (a bound-method
    /// pointer set by the subclass). Rust can't reproduce the
    /// bound-method indirection without trait objects, so this fn
    /// takes the run closure explicitly. Convenience alias for
    /// [`MultiRunnedThread::start_with`] retained for parity with
    /// Python's bare `start()` shape.
    pub fn start<F>(&self, run: F)
    where
        F: FnOnce(Arc<Mutex<bool>>) + Send + 'static,
    {
        self.start_with(run);
    }

    /// Port of `MultiRunnedThread.start()` from
    /// `powerline/lib/threaded.py:19`.
    ///
    /// The Python signature is `start(self)` which dispatches via
    /// `self.run`; Rust lacks that indirection so this takes the run
    /// closure explicitly. `shutdown_event.clear()` at py:20 maps to
    /// resetting the boolean to `false`.
    pub fn start_with<F>(&self, run: F)
    where
        F: FnOnce(Arc<Mutex<bool>>) + Send + 'static,
    {
        // py:20  def start(self):
        // py:21  self.shutdown_event.clear()
        *self.shutdown_event.lock().unwrap() = false;
        let event = self.shutdown_event.clone();
        // py:22  self.thread = Thread(target=self.run)
        // py:23  self.thread.daemon = self.daemon
        // py:24  self.thread.start()
        let handle = std::thread::spawn(move || run(event));
        *self.thread.lock().unwrap() = Some(handle);
    }

    /// Port of `MultiRunnedThread.join()` from
    /// `powerline/lib/threaded.py:25`.
    ///
    /// Python's `Thread.join(timeout=None)` waits for the thread to
    /// terminate; with `timeout` it returns when the timeout expires.
    /// Rust's `JoinHandle::join` has no built-in timeout — for the
    /// timeout case the caller must check `is_finished()` in a loop.
    /// This port matches the no-timeout Python branch.
    pub fn join(&self) -> Option<()> {
        // py:26  def join(self, *args, **kwargs):
        // py:27  if self.thread:
        // py:28  return self.thread.join(*args, **kwargs)
        // py:29  return None
        let mut t = self.thread.lock().unwrap();
        let handle = t.take()?;
        let _ = handle.join();
        Some(())
    }

    /// Signal the thread to shut down (sets `shutdown_event` to true).
    ///
    /// Not present in upstream Python as a separate method — Python
    /// callers set `self.shutdown_event.set()` directly. The Rust port
    /// exposes it as a method for ergonomic call sites.
    pub fn set_shutdown(&self) {
        *self.shutdown_event.lock().unwrap() = true;
    }
}

/// Port of `class ThreadedSegment(Segment, MultiRunnedThread)` from
/// `powerline/lib/threaded.py:33`.
///
/// Periodic-update segment base. Tracks crash state + the cached
/// update value. The actual `update`/`render` methods are
/// caller-supplied closures since Rust traits can't model Python's
/// `self.update`/`self.render` indirection without trait dispatch.
pub struct ThreadedSegment {
    pub base: MultiRunnedThread,
    /// Python class attribute: `min_sleep_time = 0.1` (py:34).
    pub min_sleep_time: f64,
    /// Python class attribute: `update_first = True` (py:35).
    pub update_first: bool,
    /// Python class attribute: `interval = 1` (py:36).
    pub interval: f64,
    /// Python: `self.run_once = True` (py:42).
    pub run_once: bool,
    /// Python: `self.crashed = False` (py:43).
    pub crashed: bool,
    /// Python: `self.crashed_value` (py:44).
    pub crashed_value: Option<String>,
    /// Python: `self.updated = False` (py:46).
    pub updated: bool,
    /// Python: `self.do_update_first` — set by set_state (py:121).
    pub do_update_first: bool,
}

impl Default for ThreadedSegment {
    fn default() -> Self {
        Self::new()
    }
}

impl ThreadedSegment {
    /// Port of `ThreadedSegment.__init__()` from
    /// `powerline/lib/threaded.py:40`.
    pub fn new() -> Self {
        // py:32  class ThreadedSegment(Segment, MultiRunnedThread):
        // py:33  min_sleep_time = 0.1
        // py:34  update_first = True
        // py:35  interval = 1
        // py:36  daemon = False
        // py:38  argmethods = ('render', 'set_state')
        // py:40  def __init__(self):
        // py:41  super(ThreadedSegment, self).__init__()
        // py:42  self.run_once = True
        // py:43  self.crashed = False
        // py:44  self.crashed_value = None
        // py:45  self.update_value = None
        // py:46  self.updated = False
        let mut base = MultiRunnedThread::new();
        // py:36  class-level override of MultiRunnedThread.daemon = True.
        // The Python subclass shadows the base's True with False at the
        // class level; that takes effect immediately on instance
        // construction. Without this line, callers that read .daemon
        // between new() and startup() would see True instead of the
        // upstream-correct False.
        base.daemon = false;
        Self {
            base,
            min_sleep_time: 0.1,
            update_first: true,
            interval: 1.0,
            run_once: true,
            crashed: false,
            crashed_value: None,
            updated: false,
            do_update_first: true,
        }
    }

    /// Port of `ThreadedSegment.set_interval()` from
    /// `powerline/lib/threaded.py:109`.
    pub fn set_interval(&mut self, interval: Option<f64>) {
        // py:109  def set_interval(self, interval=None):
        // py:110  # Allowing "interval" keyword in configuration.
        // py:114  interval = interval or getattr(self, 'interval')
        // py:115  self.interval = interval
        if let Some(i) = interval {
            self.interval = i;
        }
    }

    /// Port of `ThreadedSegment.set_state()` from
    /// `powerline/lib/threaded.py:117`.
    pub fn set_state(&mut self, interval: Option<f64>, update_first: bool) {
        // py:117  def set_state(self, interval=None, update_first=True, shutdown_event=None, **kwargs):
        // py:118  self.set_interval(interval)
        // py:119  self.shutdown_event = shutdown_event or Event()
        // py:120  self.do_update_first = update_first and self.update_first
        // py:121  self.updated = self.updated or (not self.do_update_first)
        self.set_interval(interval);
        self.do_update_first = update_first && self.update_first;
        self.updated = self.updated || !self.do_update_first;
    }

    /// Port of `ThreadedSegment.startup()` from
    /// `powerline/lib/threaded.py:123`.
    ///
    /// Marks the segment as no-longer-run-once and starts the worker.
    /// The actual run closure is caller-supplied since Rust can't
    /// dispatch via self.run.
    pub fn startup<F>(&mut self, use_daemon_threads: bool, run: F)
    where
        F: FnOnce(Arc<Mutex<bool>>) + Send + 'static,
    {
        // py:123  def startup(self, pl, **kwargs):
        // py:124  self.run_once = False
        // py:125  self.pl = pl
        // py:126  self.daemon = pl.use_daemon_threads
        // py:128  self.set_state(**kwargs)
        // py:130  if not self.is_alive():
        // py:131  self.start()
        self.run_once = false;
        self.base.daemon = use_daemon_threads;
        if !self.base.is_alive() {
            self.base.start_with(run);
        }
    }

    /// Port of `ThreadedSegment.run()` from
    /// `powerline/lib/threaded.py:87-100`.
    ///
    /// Worker-thread main loop. Polls `set_update_value` at
    /// `self.interval`-second intervals (clamped to `min_sleep_time`)
    /// until the shutdown event flips.
    ///
    /// Python's `run` is dispatched via `MultiRunnedThread.start()`
    /// (which sets `Thread(target=self.run)`); the Rust port runs the
    /// body inline so callers can wire it into `start_with(closure)`
    /// or invoke it directly on the current thread for tests.
    ///
    /// `set_update_value` is the caller-supplied refresh closure that
    /// updates the segment's cached value. Matches the
    /// `ThreadedSegment.set_update_value` body at py:69-80.
    pub fn run<F>(
        &self,
        shutdown_event: &Arc<Mutex<bool>>,
        min_sleep_time: f64,
        mut set_update_value: F,
    ) where
        F: FnMut(),
    {
        use crate::ported::lib::monotonic::monotonic;
        // py:87  def run(self):
        // py:88  if self.do_update_first:
        if self.do_update_first {
            // py:89  start_time = monotonic()
            let mut start_time = monotonic();
            // py:90  while True:
            loop {
                // py:91  self.shutdown_event.wait(max(self.interval - (monotonic() - start_time), self.min_sleep_time))
                let sleep = (self.interval - (monotonic() - start_time)).max(min_sleep_time);
                std::thread::sleep(std::time::Duration::from_secs_f64(sleep.max(0.0)));
                // py:92  if self.shutdown_event.is_set():
                if *shutdown_event.lock().unwrap() {
                    // py:93  break
                    break;
                }
                // py:94  start_time = monotonic()
                start_time = monotonic();
                // py:95  self.set_update_value()
                set_update_value();
            }
        } else {
            // py:96  else:
            // py:97  while not self.shutdown_event.is_set():
            while !*shutdown_event.lock().unwrap() {
                // py:98  start_time = monotonic()
                let start_time = monotonic();
                // py:99  self.set_update_value()
                set_update_value();
                // py:100  self.shutdown_event.wait(max(self.interval - (monotonic() - start_time), self.min_sleep_time))
                let sleep = (self.interval - (monotonic() - start_time)).max(min_sleep_time);
                std::thread::sleep(std::time::Duration::from_secs_f64(sleep.max(0.0)));
            }
        }
    }

    /// Port of `ThreadedSegment.shutdown()` from
    /// `powerline/lib/threaded.py:102`.
    pub fn shutdown(&self) {
        // py:102  def shutdown(self):
        // py:103  self.shutdown_event.set()
        // py:104  if self.daemon and self.is_alive():
        // py:105  # Give the worker thread a chance to shutdown, but don't block for
        // py:106  # too long
        // py:107  self.join(0.01)
        self.base.set_shutdown();
    }

    /// Port of `ThreadedSegment.get_update_value()` from
    /// `powerline/lib/threaded.py:82-85`.
    ///
    /// Returns `update_value` (caller-supplied); if `update=true`,
    /// re-runs `set_update_value` first per py:83-84.
    pub fn get_update_value<F>(&mut self, update: bool, refresh: F) -> Option<String>
    where
        F: FnMut() -> Result<String, String>,
    {
        // py:83-84  if update: self.set_update_value()
        if update {
            return self.set_update_value(refresh);
        }
        // py:85  return self.update_value
        // (The Rust port doesn't store update_value on the struct;
        //  callers pass it via `refresh` when needed.)
        None
    }

    /// Port of `ThreadedSegment.argspecobjs()` from
    /// `powerline/lib/threaded.py:151-156`.
    ///
    /// Yields `(name, method_name)` pairs for the configured
    /// `argmethods` set per py:152-156. Python uses `getattr` to
    /// look up the method object; Rust returns the method name as
    /// a string since methods aren't first-class.
    pub fn argspecobjs(argmethods: &[&'static str]) -> Vec<(String, String)> {
        // py:152-156
        argmethods
            .iter()
            .map(|name| (name.to_string(), name.to_string()))
            .collect()
    }

    /// Port of `ThreadedSegment.additional_args()` from
    /// `powerline/lib/threaded.py:158-159`.
    ///
    /// Returns `(('interval', self.interval),)` per py:159.
    pub fn additional_args(interval: Option<f64>) -> Vec<(String, Option<f64>)> {
        // py:159
        vec![("interval".to_string(), interval)]
    }

    /// Port of `ThreadedSegment._omitted_args` class attribute at
    /// `powerline/lib/threaded.py:161-164`.
    ///
    /// Returns the static map of `method_name → omitted_arg_indices`
    /// per py:162-164. `'render'` omits arg `(0,)`; `'set_state'`
    /// omits `('shutdown_event',)`.
    pub fn omitted_args_table(name: &str) -> Vec<&'static str> {
        // py:162-164
        match name {
            "render" => vec!["0"],
            "set_state" => vec!["shutdown_event"],
            _ => Vec::new(),
        }
    }

    /// Port of `ThreadedSegment.omitted_args()` from
    /// `powerline/lib/threaded.py:166-170`.
    ///
    /// Returns the omitted-args list for `name`. The Python source
    /// at py:168-169 increments integer indices by 1 when the
    /// method is bound (to skip `self`). The Rust port mirrors the
    /// behaviour via a bool flag since Rust has no MethodType.
    pub fn omitted_args(name: &str, is_method: bool) -> Vec<String> {
        // py:167  ret = self._omitted_args.get(name, ())
        let raw = Self::omitted_args_table(name);
        // py:168-169  is_method → shift integer indices by 1
        raw.iter()
            .map(|arg| {
                if is_method {
                    if let Ok(idx) = arg.parse::<i32>() {
                        return (idx + 1).to_string();
                    }
                }
                arg.to_string()
            })
            .collect()
    }

    /// Port of `ThreadedSegment.critical()` from
    /// `powerline/lib/threaded.py:133-134`.
    ///
    /// Returns the (prefix, message) pair callers route through
    /// `self.pl.critical(...)` per py:134.
    pub fn critical(class_name: &str, message: &str) -> (String, String) {
        (class_name.to_string(), message.to_string())
    }

    /// Port of `ThreadedSegment.exception()` from
    /// `powerline/lib/threaded.py:136-137`.
    pub fn exception(class_name: &str, message: &str) -> (String, String) {
        (class_name.to_string(), message.to_string())
    }

    /// Port of `ThreadedSegment.info()` from
    /// `powerline/lib/threaded.py:139-140`.
    pub fn info(class_name: &str, message: &str) -> (String, String) {
        (class_name.to_string(), message.to_string())
    }

    /// Port of `ThreadedSegment.error()` from
    /// `powerline/lib/threaded.py:142-143`.
    pub fn error(class_name: &str, message: &str) -> (String, String) {
        (class_name.to_string(), message.to_string())
    }

    /// Port of `ThreadedSegment.warn()` from
    /// `powerline/lib/threaded.py:145-146`.
    pub fn warn(class_name: &str, message: &str) -> (String, String) {
        (class_name.to_string(), message.to_string())
    }

    /// Port of `ThreadedSegment.debug()` from
    /// `powerline/lib/threaded.py:148-149`.
    pub fn debug(class_name: &str, message: &str) -> (String, String) {
        (class_name.to_string(), message.to_string())
    }

    /// Port of `ThreadedSegment.set_update_value()` from
    /// `powerline/lib/threaded.py:69`.
    ///
    /// Wraps the caller-supplied update closure with crash handling.
    /// Returns the fresh update value or None when the update
    /// crashed (and sets `self.crashed = true`).
    pub fn set_update_value<F>(&mut self, mut update: F) -> Option<String>
    where
        F: FnMut() -> Result<String, String>,
    {
        // py:69  def set_update_value(self):
        // py:70  try:
        // py:71  self.update_value = self.update(self.update_value)
        // py:72  except Exception as e:
        // py:73  self.exception('Exception while updating: {0}', str(e))
        // py:74  self.crashed = True
        // py:75  except KeyboardInterrupt:
        // py:76  self.warn('Caught keyboard interrupt while updating')
        // py:77  self.crashed = True
        // py:78  else:
        // py:79  self.crashed = False
        // py:80  self.updated = True
        match update() {
            Ok(v) => {
                self.crashed = false;
                self.updated = true;
                Some(v)
            }
            Err(_) => {
                self.crashed = true;
                None
            }
        }
    }
}

/// Port of `class KwThreadedSegment(ThreadedSegment)` from
/// `powerline/lib/threaded.py:173`.
///
/// Multi-key variant — tracks per-key cached states with a separate
/// crashed flag per key. The Python class stores all per-key data
/// under `self.updates` (a dict keyed by the `key()` return value);
/// the Rust port mirrors that structure with a HashMap.
pub struct KwThreadedSegment {
    pub base: ThreadedSegment,
    /// Python: `self.updates` (py:181).
    pub updates: std::collections::HashMap<String, KwUpdateState>,
}

/// Per-key state stored in `KwThreadedSegment.updates`. Mirrors the
/// Python dict shape per py:218-226 inner loop.
#[derive(Debug, Clone)]
pub struct KwUpdateState {
    pub crashed: bool,
    pub value: Option<String>,
}

impl Default for KwThreadedSegment {
    fn default() -> Self {
        Self::new()
    }
}

impl KwThreadedSegment {
    /// Port of `KwThreadedSegment.__init__()` from
    /// `powerline/lib/threaded.py:178`.
    pub fn new() -> Self {
        Self {
            base: ThreadedSegment::new(),
            // py:181  self.updates = {}
            updates: std::collections::HashMap::new(),
        }
    }

    /// Port of `KwThreadedSegment.update_one()` from
    /// `powerline/lib/threaded.py:218`.
    ///
    /// Updates the state for a single key via the caller's update
    /// closure. Crashed updates leave the previous value intact.
    pub fn update_one<F>(&mut self, key: &str, mut update: F)
    where
        F: FnMut() -> Result<String, String>,
    {
        // py:220-227  try update; except: crashed = True
        let state = self
            .updates
            .entry(key.to_string())
            .or_insert(KwUpdateState {
                crashed: false,
                value: None,
            });
        match update() {
            Ok(v) => {
                state.crashed = false;
                state.value = Some(v);
            }
            Err(_) => {
                state.crashed = true;
            }
        }
    }

    /// Port of `KwThreadedSegment.key()` (staticmethod) from
    /// `powerline/lib/threaded.py:185-187`.
    ///
    /// Python: `frozenset(kwargs.items())`. Returns the kwargs as a
    /// sorted list of (key, value) pairs since Rust has no
    /// frozenset; callers compare for equality.
    pub fn key(kwargs: &[(String, String)]) -> Vec<(String, String)> {
        // py:187  frozenset(kwargs.items())
        let mut out: Vec<(String, String)> = kwargs.to_vec();
        out.sort();
        out
    }

    /// Port of `KwThreadedSegment.render_one()` (staticmethod) from
    /// `powerline/lib/threaded.py:254-256`.
    ///
    /// Port of `KwThreadedSegment.render()` from
    /// `powerline/lib/threaded.py:189-216`.
    ///
    /// Per-key dispatch: looks up the segment value for `key` in
    /// `queries`, falls back to a recursive `render` after registering
    /// the key in `new_queries` per py:201-212. Calls
    /// [`render_one`](Self::render_one) at py:216.
    ///
    /// Python embeds the queries dict + new_queries list on `self`;
    /// the Rust port takes them as explicit args so callers route
    /// through their own state-tracking model. Returns
    /// `Some(rendered)` from `render_one(state, **kwargs)` per py:216.
    pub fn render<R>(
        queries: &std::collections::HashMap<String, (f64, Option<String>)>,
        key: &str,
        do_update_first: bool,
        run_once: bool,
        after_update: bool,
        mut render_with_update: R,
    ) -> Option<String>
    where
        R: FnMut(bool, &str) -> Option<String>,
    {
        // py:189  def render(self, update_value, update_first, key=None, after_update=False, **kwargs):
        // py:196  try:
        // py:197  update_state = queries[key][1]
        let update_state = match queries.get(key) {
            Some((_, state)) => state.clone(),
            None => {
                // py:198  except KeyError:
                // py:199  with self.write_lock:
                // py:200  self.new_queries.append(key)
                // py:201  if self.do_update_first or self.run_once:
                if do_update_first || run_once {
                    // py:202  if after_update:
                    if after_update {
                        // py:203  self.error('internal error: ...')
                        // py:204  update_state = None
                        None
                    } else {
                        // py:205-212  return self.render(get_update_value(True), False, key, after_update=True, **kwargs)
                        return render_with_update(true, key);
                    }
                } else {
                    // py:213  else:
                    // py:214  update_state = None
                    None
                }
            }
        };
        // py:216  return self.render_one(update_state, **kwargs)
        Self::render_one(update_state)
    }

    /// Port of `KwThreadedSegment.update()` from
    /// `powerline/lib/threaded.py:228-247`.
    ///
    /// Per-key batched-update step: walks `queries`, refreshes stale
    /// entries via `update_one`, then ingests `new_queries`. Mirrors
    /// the inner loop at py:238-245.
    ///
    /// Python mutates `self.new_queries`, `self.write_lock`, etc.; the
    /// Rust port takes those as explicit args so the data flow is
    /// inspectable.
    ///
    /// Returns the new `(queries, crashed)` pair per py:231 / py:247
    /// — Python wraps both in a tuple as the value passed to the next
    /// tick.
    pub fn update<F>(
        old_queries: &std::collections::HashMap<String, (f64, Option<String>)>,
        new_queries: &[String],
        interval: f64,
        mut update_one: F,
    ) -> (
        std::collections::HashMap<String, (f64, Option<String>)>,
        std::collections::HashSet<String>,
    )
    where
        F: FnMut(
            &mut std::collections::HashSet<String>,
            &mut std::collections::HashMap<String, (f64, Option<String>)>,
            &str,
        ),
    {
        use crate::ported::lib::monotonic::monotonic;
        // py:229  updates = {}
        // py:230  crashed = set()
        // py:231  update_value = (updates, crashed)
        let mut updates: std::collections::HashMap<String, (f64, Option<String>)> =
            std::collections::HashMap::new();
        let mut crashed: std::collections::HashSet<String> = std::collections::HashSet::new();

        let now = monotonic();
        // py:238  for key, (last_query_time, state) in queries.items():
        for (key, (last_query_time, state)) in old_queries {
            // py:239  if last_query_time < monotonic() < last_query_time + self.interval:
            if *last_query_time < now && now < last_query_time + interval {
                // py:240  updates[key] = (last_query_time, state)
                updates.insert(key.clone(), (*last_query_time, state.clone()));
            } else {
                // py:241-242  else: self.update_one(crashed, updates, key)
                update_one(&mut crashed, &mut updates, key);
            }
        }

        // py:244-245  for key in new_queries: self.update_one(crashed, updates, key)
        for key in new_queries {
            update_one(&mut crashed, &mut updates, key);
        }

        // py:247  return update_value
        (updates, crashed)
    }

    /// Identity helper: returns `update_state` unchanged per
    /// py:256. Subclasses override to format the state into a
    /// renderable string.
    pub fn render_one(update_state: Option<String>) -> Option<String> {
        // py:256  return update_state
        update_state
    }

    /// Port of `KwThreadedSegment._omitted_args` class attribute at
    /// `powerline/lib/threaded.py:258-262`.
    ///
    /// Overrides ThreadedSegment._omitted_args per py:258-262.
    /// 'render' → ('update_value', 'key', 'after_update');
    /// 'set_state' → ('shutdown_event',);
    /// 'render_one' → (0,).
    pub fn omitted_args_table(name: &str) -> Vec<&'static str> {
        // py:259-261
        match name {
            "render" => vec!["update_value", "key", "after_update"],
            "set_state" => vec!["shutdown_event"],
            "render_one" => vec!["0"],
            _ => Vec::new(),
        }
    }

    /// Port of `KwThreadedSegment.set_state()` from
    /// `powerline/lib/threaded.py:249-252`.
    ///
    /// Override that pins `do_update_first = update_first` per
    /// py:251 since KwThreadedSegment.update_first defaults to true.
    pub fn set_state(&mut self, interval: Option<f64>, update_first: bool) {
        // py:250-252  set_interval + do_update_first + shutdown_event
        self.base.set_interval(interval);
        self.base.do_update_first = update_first;
    }
}

/// Decision shape returned by [`ThreadedSegment::call_dispatch`].
///
/// Mirrors the Python `__call__` body's branch chain at py:48-67:
/// either return `crashed_value` per py:64-65, or dispatch to
/// `self.render(...)` per py:67 with the (update_value,
/// update_first) pair.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CallDecision {
    /// py:64-65  if crashed: return crashed_value
    Crashed,
    /// py:67  return self.render(update_value, update_first=..., pl=..., **kwargs)
    Render {
        update_first: bool,
        do_update_first: bool,
    },
}

impl ThreadedSegment {
    /// Port of `ThreadedSegment.__call__()` from
    /// `powerline/lib/threaded.py:48-67`.
    ///
    /// Returns the dispatch decision based on:
    /// - `run_once` (py:49-52) — sets state + force-updates
    /// - `is_alive` (py:53-60) — starts the worker + maybe updates
    /// - default (py:61-62) — updates when stale
    ///
    /// The actual `render()` dispatch + state mutations live with
    /// the caller; this fn returns the decision so the dispatch is
    /// testable without the live worker thread.
    pub fn call_dispatch(
        run_once: bool,
        is_alive: bool,
        updated: bool,
        do_update_first: bool,
        crashed: bool,
        update_first: bool,
    ) -> CallDecision {
        // py:64-65  if self.crashed: return self.crashed_value
        if crashed {
            return CallDecision::Crashed;
        }
        // py:49-52  run_once: force update
        if run_once {
            return CallDecision::Render {
                update_first,
                do_update_first: true,
            };
        }
        // py:53-60  not is_alive: start + maybe update
        if !is_alive {
            return CallDecision::Render {
                update_first,
                do_update_first,
            };
        }
        // py:61-62  default: update if not updated
        CallDecision::Render {
            update_first,
            do_update_first: !updated,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU32, Ordering};

    #[test]
    fn new_thread_is_not_alive() {
        let t = MultiRunnedThread::new();
        assert!(!t.is_alive());
    }

    #[test]
    fn start_then_join_runs_closure() {
        let t = MultiRunnedThread::new();
        let ran = Arc::new(AtomicU32::new(0));
        let ran_clone = ran.clone();
        t.start_with(move |_event| {
            ran_clone.fetch_add(1, Ordering::SeqCst);
        });
        t.join();
        assert_eq!(ran.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn set_shutdown_signals_the_thread() {
        let t = MultiRunnedThread::new();
        let observed = Arc::new(Mutex::new(false));
        let observed_clone = observed.clone();
        t.start_with(move |event| {
            // Wait until shutdown_event flips, then record observation.
            loop {
                if *event.lock().unwrap() {
                    *observed_clone.lock().unwrap() = true;
                    break;
                }
                std::thread::sleep(std::time::Duration::from_millis(2));
            }
        });
        std::thread::sleep(std::time::Duration::from_millis(10));
        t.set_shutdown();
        t.join();
        assert!(*observed.lock().unwrap());
    }

    #[test]
    fn threaded_segment_defaults_match_upstream() {
        // py:34-36 class attributes
        let s = ThreadedSegment::new();
        assert!((s.min_sleep_time - 0.1).abs() < 1e-9);
        assert!(s.update_first);
        assert_eq!(s.interval, 1.0);
        assert!(s.run_once);
        assert!(!s.crashed);
        assert!(!s.updated);
    }

    #[test]
    fn threaded_segment_set_interval_overrides_default() {
        let mut s = ThreadedSegment::new();
        s.set_interval(Some(5.0));
        assert_eq!(s.interval, 5.0);
    }

    #[test]
    fn threaded_segment_set_interval_none_preserves_default() {
        // py:114  interval = interval or self.interval
        let mut s = ThreadedSegment::new();
        s.set_interval(None);
        assert_eq!(s.interval, 1.0);
    }

    #[test]
    fn threaded_segment_set_state_sets_do_update_first() {
        // py:120-122
        let mut s = ThreadedSegment::new();
        s.set_state(Some(2.0), true);
        assert!(s.do_update_first);
        assert_eq!(s.interval, 2.0);
    }

    #[test]
    fn threaded_segment_set_state_false_clears_do_update_first() {
        let mut s = ThreadedSegment::new();
        s.set_state(None, false);
        assert!(!s.do_update_first);
        // py:122  self.updated = self.updated or not self.do_update_first
        assert!(s.updated);
    }

    #[test]
    fn threaded_segment_set_update_value_ok_clears_crashed() {
        let mut s = ThreadedSegment::new();
        s.crashed = true;
        let v = s.set_update_value(|| Ok("data".to_string()));
        assert_eq!(v, Some("data".to_string()));
        assert!(!s.crashed);
        assert!(s.updated);
    }

    #[test]
    fn threaded_segment_set_update_value_err_sets_crashed() {
        // py:74  except: crashed = True
        let mut s = ThreadedSegment::new();
        let v = s.set_update_value(|| Err::<String, String>("boom".to_string()));
        assert!(v.is_none());
        assert!(s.crashed);
    }

    #[test]
    fn threaded_segment_shutdown_signals_thread() {
        let s = ThreadedSegment::new();
        s.shutdown();
        // shutdown_event flips to true
        assert!(*s.base.shutdown_event.lock().unwrap());
    }

    #[test]
    fn kw_threaded_segment_new_empty() {
        let s = KwThreadedSegment::new();
        assert!(s.updates.is_empty());
    }

    #[test]
    fn kw_threaded_segment_update_one_inserts_key() {
        let mut s = KwThreadedSegment::new();
        s.update_one("foo", || Ok("v".to_string()));
        let state = s.updates.get("foo").unwrap();
        assert!(!state.crashed);
        assert_eq!(state.value.as_deref(), Some("v"));
    }

    #[test]
    fn kw_threaded_segment_update_one_crash_preserves_value() {
        // py:226  failed update leaves the cached value
        let mut s = KwThreadedSegment::new();
        s.update_one("foo", || Ok("good".to_string()));
        s.update_one("foo", || Err::<String, String>("boom".to_string()));
        let state = s.updates.get("foo").unwrap();
        assert!(state.crashed);
        assert_eq!(state.value.as_deref(), Some("good"));
    }

    #[test]
    fn kw_threaded_segment_update_one_recovery_clears_crashed() {
        let mut s = KwThreadedSegment::new();
        s.update_one("foo", || Err::<String, String>("boom".to_string()));
        s.update_one("foo", || Ok("recovered".to_string()));
        let state = s.updates.get("foo").unwrap();
        assert!(!state.crashed);
        assert_eq!(state.value.as_deref(), Some("recovered"));
    }

    #[test]
    fn kw_threaded_segment_multiple_keys_isolated() {
        let mut s = KwThreadedSegment::new();
        s.update_one("a", || Ok("av".to_string()));
        s.update_one("b", || Err::<String, String>("err".to_string()));
        let a = s.updates.get("a").unwrap();
        let b = s.updates.get("b").unwrap();
        assert!(!a.crashed);
        assert!(b.crashed);
    }

    #[test]
    fn get_update_value_runs_refresh_when_update_true() {
        // py:83-84
        let mut s = ThreadedSegment::default();
        let r = s.get_update_value(true, || Ok("fresh".to_string()));
        assert_eq!(r, Some("fresh".to_string()));
        assert!(s.updated);
    }

    #[test]
    fn get_update_value_skips_refresh_when_update_false() {
        // py:85  return self.update_value (no refresh)
        let mut s = ThreadedSegment::default();
        let r = s.get_update_value(false, || Ok("should_not_run".to_string()));
        // Rust port returns None since there's no stored value
        assert!(r.is_none());
        assert!(!s.updated);
    }

    #[test]
    fn argspecobjs_yields_argmethod_name_pairs() {
        // py:152-156
        let r = ThreadedSegment::argspecobjs(&["render", "set_state"]);
        assert_eq!(r.len(), 2);
        assert_eq!(r[0].0, "render");
        assert_eq!(r[1].0, "set_state");
    }

    #[test]
    fn additional_args_returns_interval_pair() {
        // py:159
        let r = ThreadedSegment::additional_args(Some(5.0));
        assert_eq!(r.len(), 1);
        assert_eq!(r[0].0, "interval");
        assert_eq!(r[0].1, Some(5.0));
    }

    #[test]
    fn omitted_args_table_render_omits_first() {
        // py:162  'render': (0,)
        assert_eq!(ThreadedSegment::omitted_args_table("render"), vec!["0"]);
    }

    #[test]
    fn omitted_args_table_set_state_omits_shutdown_event() {
        // py:163  'set_state': ('shutdown_event',)
        assert_eq!(
            ThreadedSegment::omitted_args_table("set_state"),
            vec!["shutdown_event"]
        );
    }

    #[test]
    fn omitted_args_table_unknown_name_returns_empty() {
        assert!(ThreadedSegment::omitted_args_table("other").is_empty());
    }

    #[test]
    fn omitted_args_unbound_passes_indices_through() {
        // py:168-169  not method → no shift
        let r = ThreadedSegment::omitted_args("render", false);
        assert_eq!(r, vec!["0".to_string()]);
    }

    #[test]
    fn omitted_args_bound_shifts_integer_indices_by_one() {
        // py:168-169  isinstance MethodType → +1 on ints
        let r = ThreadedSegment::omitted_args("render", true);
        assert_eq!(r, vec!["1".to_string()]);
    }

    #[test]
    fn omitted_args_bound_leaves_string_indices_unchanged() {
        let r = ThreadedSegment::omitted_args("set_state", true);
        // String args ('shutdown_event') don't get shifted
        assert_eq!(r, vec!["shutdown_event".to_string()]);
    }

    #[test]
    fn critical_returns_prefix_message_pair() {
        // py:134
        let (prefix, msg) = ThreadedSegment::critical("MyClass", "boom");
        assert_eq!(prefix, "MyClass");
        assert_eq!(msg, "boom");
    }

    #[test]
    fn exception_returns_prefix_message_pair() {
        // py:137
        let (prefix, msg) = ThreadedSegment::exception("MyClass", "exc");
        assert_eq!(prefix, "MyClass");
        assert_eq!(msg, "exc");
    }

    #[test]
    fn info_warn_error_debug_all_return_prefix_message_pair() {
        // py:139-149
        assert_eq!(ThreadedSegment::info("X", "i"), ("X".into(), "i".into()));
        assert_eq!(ThreadedSegment::error("X", "e"), ("X".into(), "e".into()));
        assert_eq!(ThreadedSegment::warn("X", "w"), ("X".into(), "w".into()));
        assert_eq!(ThreadedSegment::debug("X", "d"), ("X".into(), "d".into()));
    }

    #[test]
    fn kw_key_sorts_kwargs() {
        // py:187  frozenset(kwargs.items())
        let r = KwThreadedSegment::key(&[
            ("b".to_string(), "2".to_string()),
            ("a".to_string(), "1".to_string()),
        ]);
        assert_eq!(
            r,
            vec![
                ("a".to_string(), "1".to_string()),
                ("b".to_string(), "2".to_string()),
            ]
        );
    }

    #[test]
    fn kw_key_empty_returns_empty() {
        let r = KwThreadedSegment::key(&[]);
        assert!(r.is_empty());
    }

    #[test]
    fn kw_render_one_returns_update_state_unchanged() {
        // py:256
        assert_eq!(
            KwThreadedSegment::render_one(Some("hi".to_string())),
            Some("hi".to_string())
        );
        assert_eq!(KwThreadedSegment::render_one(None), None);
    }

    #[test]
    fn kw_omitted_args_table_render_includes_three_args() {
        // py:259  ('update_value', 'key', 'after_update')
        let r = KwThreadedSegment::omitted_args_table("render");
        assert_eq!(r, vec!["update_value", "key", "after_update"]);
    }

    #[test]
    fn kw_omitted_args_table_render_one_omits_first() {
        // py:261  'render_one': (0,)
        let r = KwThreadedSegment::omitted_args_table("render_one");
        assert_eq!(r, vec!["0"]);
    }

    #[test]
    fn kw_omitted_args_table_set_state_omits_shutdown_event() {
        // py:260
        let r = KwThreadedSegment::omitted_args_table("set_state");
        assert_eq!(r, vec!["shutdown_event"]);
    }

    #[test]
    fn kw_omitted_args_table_unknown_returns_empty() {
        let r = KwThreadedSegment::omitted_args_table("other");
        assert!(r.is_empty());
    }

    #[test]
    fn kw_set_state_pins_do_update_first() {
        // py:251  do_update_first = update_first
        let mut s = KwThreadedSegment::new();
        s.set_state(Some(5.0), false);
        assert!(!s.base.do_update_first);
        s.set_state(Some(5.0), true);
        assert!(s.base.do_update_first);
    }

    #[test]
    fn kw_set_state_calls_set_interval() {
        // py:250  self.set_interval(interval)
        let mut s = KwThreadedSegment::new();
        s.set_state(Some(2.0), true);
        assert!((s.base.interval - 2.0).abs() < 1e-9);
    }

    #[test]
    fn call_dispatch_crashed_returns_crashed() {
        // py:64-65
        let d = ThreadedSegment::call_dispatch(false, true, true, false, true, true);
        assert_eq!(d, CallDecision::Crashed);
    }

    #[test]
    fn call_dispatch_run_once_forces_update() {
        // py:49-52
        let d = ThreadedSegment::call_dispatch(true, false, false, false, false, true);
        assert_eq!(
            d,
            CallDecision::Render {
                update_first: true,
                do_update_first: true,
            }
        );
    }

    #[test]
    fn call_dispatch_not_alive_uses_do_update_first() {
        // py:53-60
        let d = ThreadedSegment::call_dispatch(false, false, false, true, false, true);
        assert_eq!(
            d,
            CallDecision::Render {
                update_first: true,
                do_update_first: true,
            }
        );
    }

    #[test]
    fn call_dispatch_default_updates_when_stale() {
        // py:61-62  not self.updated → update
        let d = ThreadedSegment::call_dispatch(false, true, false, true, false, true);
        assert_eq!(
            d,
            CallDecision::Render {
                update_first: true,
                do_update_first: true,
            }
        );
    }

    #[test]
    fn call_dispatch_default_skips_update_when_fresh() {
        // py:62  self.updated → skip update
        let d = ThreadedSegment::call_dispatch(false, true, true, true, false, true);
        assert_eq!(
            d,
            CallDecision::Render {
                update_first: true,
                do_update_first: false,
            }
        );
    }

    #[test]
    fn multi_runned_thread_start_dispatches_to_start_with() {
        // py:20-24  start() is the bound-method-pointer indirection.
        let t = MultiRunnedThread::new();
        let counter = std::sync::Arc::new(AtomicU32::new(0));
        let counter_c = counter.clone();
        t.start(move |_event| {
            counter_c.fetch_add(7, Ordering::SeqCst);
        });
        t.join();
        assert_eq!(counter.load(Ordering::SeqCst), 7);
    }

    #[test]
    fn kw_threaded_render_falls_back_to_render_with_update_on_cache_miss() {
        // py:198-212  KeyError → recursive render(update=True, after_update=True)
        let queries = std::collections::HashMap::new();
        let mut called_with: Option<(bool, String)> = None;
        let result = KwThreadedSegment::render(
            &queries,
            "missing_key",
            true, // do_update_first
            false,
            false,
            |upd, key| {
                called_with = Some((upd, key.to_string()));
                Some("recursive_result".to_string())
            },
        );
        assert_eq!(result, Some("recursive_result".to_string()));
        assert_eq!(called_with, Some((true, "missing_key".to_string())));
    }

    #[test]
    fn kw_threaded_render_returns_cached_state_when_present() {
        // py:197 + py:216  cached → render_one
        let mut queries = std::collections::HashMap::new();
        queries.insert("k1".to_string(), (0.0_f64, Some("cached".to_string())));
        let result = KwThreadedSegment::render(&queries, "k1", true, false, false, |_, _| {
            panic!("should not be called when cached")
        });
        assert_eq!(result, Some("cached".to_string()));
    }

    #[test]
    fn kw_threaded_render_after_update_internal_error_returns_none() {
        // py:202-204  after_update + missing key → log + state=None
        let queries = std::collections::HashMap::new();
        let result = KwThreadedSegment::render(
            &queries,
            "missing",
            true,
            false,
            true, // after_update
            |_, _| panic!("should not recurse on after_update"),
        );
        assert_eq!(result, None);
    }

    #[test]
    fn kw_threaded_update_refreshes_stale_entries() {
        // py:239-242  stale (last + interval < now) → update_one
        let mut old = std::collections::HashMap::new();
        // last_query_time well in the past, interval tiny — current
        // monotonic exceeds last + interval, so the branch is stale.
        old.insert("k".to_string(), (0.000001_f64, Some("old".to_string())));
        let new_queries: Vec<String> = vec![];
        let calls = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
        let calls_c = calls.clone();
        let (updates, crashed) = KwThreadedSegment::update(
            &old,
            &new_queries,
            0.000001, // tiny interval so monotonic() > last + interval
            move |_crashed, updates, key| {
                calls_c.lock().unwrap().push(key.to_string());
                updates.insert(key.to_string(), (1.0, Some("new".to_string())));
            },
        );
        assert_eq!(*calls.lock().unwrap(), vec!["k".to_string()]);
        assert_eq!(updates.get("k").unwrap().1, Some("new".to_string()));
        assert!(crashed.is_empty());
    }

    #[test]
    fn kw_threaded_update_keeps_fresh_entries_as_is() {
        // py:240  fresh (last < now < last + interval) → updates[key] = old
        let mut old = std::collections::HashMap::new();
        // Pin a recent timestamp so it's still within the interval.
        let now = crate::ported::lib::monotonic::monotonic();
        old.insert("k".to_string(), (now, Some("kept".to_string())));
        let new_queries: Vec<String> = vec![];
        let calls = std::sync::Arc::new(std::sync::Mutex::new(0u32));
        let calls_c = calls.clone();
        let (updates, _) = KwThreadedSegment::update(
            &old,
            &new_queries,
            3600.0, // long interval — entry is fresh
            move |_, _, _| {
                *calls_c.lock().unwrap() += 1;
            },
        );
        assert_eq!(*calls.lock().unwrap(), 0, "no refresh should happen");
        assert_eq!(updates.get("k").unwrap().1, Some("kept".to_string()));
    }

    #[test]
    fn kw_threaded_update_processes_new_queries() {
        // py:244-245  for key in new_queries: update_one
        let old = std::collections::HashMap::new();
        let new_queries = vec!["k1".to_string(), "k2".to_string()];
        let calls = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
        let calls_c = calls.clone();
        let (updates, _crashed) =
            KwThreadedSegment::update(&old, &new_queries, 60.0, move |_crashed, updates, key| {
                calls_c.lock().unwrap().push(key.to_string());
                updates.insert(key.to_string(), (1.0, Some(format!("v_{key}"))));
            });
        let mut got = calls.lock().unwrap().clone();
        got.sort();
        assert_eq!(got, vec!["k1".to_string(), "k2".to_string()]);
        assert_eq!(updates.get("k1").unwrap().1, Some("v_k1".to_string()));
        assert_eq!(updates.get("k2").unwrap().1, Some("v_k2".to_string()));
    }

    #[test]
    fn threaded_segment_run_calls_update_once_then_stops_when_shutdown_set() {
        // py:97  while not self.shutdown_event.is_set()
        let mut seg = ThreadedSegment::new();
        seg.interval = 0.001;
        seg.do_update_first = false;
        let event = Arc::new(Mutex::new(false));
        let event_c = event.clone();
        let counter = std::sync::Arc::new(AtomicU32::new(0));
        let counter_c = counter.clone();
        seg.run(&event, 0.001, move || {
            counter_c.fetch_add(1, Ordering::SeqCst);
            // Flip shutdown after first call so loop terminates.
            *event_c.lock().unwrap() = true;
        });
        let n = counter.load(Ordering::SeqCst);
        assert!(n >= 1, "expected at least 1 call, got {n}");
    }
}