Add live.poll() and retire utils.processor

.issueflows/03-solved-issues/issue781_original.md

@@ -0,0 +1,15 @@
+# Issue #781: L4 — live.py poll loop; retire processor.py
+
+- GitHub: https://github.com/jepegit/cellpy/issues/781
+- Epic: #783 (Epic L), Stage 3. Depends on: #164.
+
+## Original description
+
+Epic L of cellpy 2.2 (Stage 5). Design: live-incremental §5. Depends on L3 (#164).
+
+- `utils/live.py`: replace the stub with
+  `poll(cell_or_path, interval, on_update=, stop_when_complete=True)` calling `c.update()`.
+- Delete `utils/processor.py` — fold its one real idea (thread-pool `cellpy.get`
+  fan-out) into `batch.runner`'s existing executor path.
+- Fix the `batch_core.py:180` `accessor_label.lstrip(self.accessor_pre)` →
+  `removeprefix(...)` bug if it still exists post-batch-v3.

.issueflows/03-solved-issues/issue781_plan.md

@@ -0,0 +1,15 @@
+# Plan: #781 `live.poll` and retire `processor.py`
+
+Autonomous run under #783 (user: "process the issues").
+
+1. `cellpy/utils/live.py`: `poll(cell_or_path, interval=30, on_update=None,
+   stop_when_complete=True, until=None, max_polls=None, timeout=None,
+   raise_errors=False, sleep=time.sleep, **get_kwargs)`; returns the cell;
+   run bookkeeping in `PollStatus` on `cell.poll_status`.
+2. `CellpyCell.source_complete` set from `IncrementalChunk.complete` in
+   `_update_incremental` so `stop_when_complete` has a signal.
+3. `git rm cellpy/utils/processor.py`; `batch.runner` already provides the
+   threads executor. `utils/batch_tools/` no longer exists → no `lstrip` fix.
+4. Tests `tests/test_live_poll.py` with a fake clock that grows the file.
+5. Docs: folder-structure page, `docs/agents/index.md`, design doc section,
+   HISTORY, test registry.

.issueflows/03-solved-issues/issue781_status.md

@@ -0,0 +1,21 @@
+# Status: #781 `live.poll` and retire `processor.py`
+
+- [x] Done
+
+## Done
+
+- `cellpy/utils/live.py`: `poll()` + `PollStatus`.
+- `CellpyCell.source_complete` (init + set in `_update_incremental`).
+- `cellpy/utils/processor.py` deleted; folder-structure doc updated.
+- `tests/test_live_poll.py`: 8 essential tests.
+- Docs: `docs/agents/index.md`, `incremental-load-protocol.md` (#781 section),
+  HISTORY, test registry.
+
+## Notes
+
+- `batch_core.py` `lstrip` bug: module no longer exists (batch v3) — nothing to fix.
+- Branch `781-live-poll` stacked on `164-cell-update` (PR #1102).
+
+## Remaining
+
+- None. #782 (batch live refresh) next.

.issueflows/04-designs-and-guides/incremental-load-protocol.md

@@ -105,12 +105,26 @@ The public consumer of the protocol, in `cellpy/readers/cellreader.py`
   delegates to `CellpyCell._update_from_raw_rows`, so the equality tests
   cover the shipped engine.
 
-## Link
+## `live.poll()` (#781)
+
+`cellpy/utils/live.py`. A loop over `update()` with injectable `sleep`
+(tests use a fake clock that grows the file on chosen ticks). Stop
+conditions, checked before each sleep: `cell.source_complete` (set by
+`update()` from `IncrementalChunk.complete`; no shipped loader sets it yet),
+`until(cell)`, `max_polls`, `timeout`. `on_update(cell)` fires only on
+ticks that changed the frames (and for the initial load when a path is
+passed). Errors from `update`/`on_update` end the loop with
+`stopped_by="error"` unless `raise_errors=True`; `KeyboardInterrupt` ends it
+with `"interrupted"`. Run bookkeeping lives on `cell.poll_status`
+(`PollStatus`). Only `model=` from the `get` kwargs is forwarded to
+`update()` (the one loader kwarg that is not persisted).
+
+`utils/processor.py` (thread-pool `cellpy.get` fan-out sketch) is deleted;
+`batch.runner` already has `executor="threads"`. The `batch_core.py`
+`lstrip` bug is moot: `utils/batch_tools/` is gone since batch v3.
 
-Design §3 in `cellpy-design-and-development/active/cellpy2-live-incremental-design.md`.
-Tests: `tests/test_load_since.py` (#780), `tests/test_cell_update.py` (#164).
-Consumers: `live.py` poll loop (#781), batch live refresh (#782).
 ## Link
 
 Design §3 in `cellpy-design-and-development/active/cellpy2-live-incremental-design.md`.
-`CellpyCell.update()` is #164. Tests: `tests/test_load_since.py`.
+Tests: `tests/test_load_since.py` (#780), `tests/test_cell_update.py` (#164),
+`tests/test_live_poll.py` (#781). Consumer: batch live refresh (#782).

.issueflows/04-designs-and-guides/test-registry.md

@@ -209,6 +209,14 @@ current issue**. `/iflow-doctor` may audit the whole suite against this table.
 | tests/test_cell_update.py::test_update_force_reloads_an_unchanged_source | yes | yes | CellpyCell.update(force=True) | #164 | |
 | tests/test_cell_update.py::test_update_without_raw_source_raises | yes | yes | CellpyCell.update | #164 | NoDataFound |
 | tests/test_cell_update.py::test_update_uses_full_reload_for_non_incremental_loader | yes | yes | CellpyCell.update (protocol gate) | #164 | loader without load_since → full reload |
+| tests/test_live_poll.py::test_poll_updates_on_growth_and_stops_at_max_polls | yes | yes | utils.live.poll | #781 | fake clock grows file; result == full load |
+| tests/test_live_poll.py::test_poll_from_path_fires_on_update_for_the_initial_load | yes | yes | utils.live.poll (path input) | #781 | |
+| tests/test_live_poll.py::test_poll_stops_on_until | yes | yes | utils.live._should_stop | #781 | |
+| tests/test_live_poll.py::test_poll_stops_when_source_complete | yes | yes | utils.live.poll / CellpyCell.source_complete | #781 | |
+| tests/test_live_poll.py::test_poll_stops_on_timeout | yes | yes | utils.live._should_stop | #781 | monotonic patched |
+| tests/test_live_poll.py::test_poll_records_update_errors_unless_raise_errors | yes | yes | utils.live.poll | #781 | |
+| tests/test_live_poll.py::test_poll_keyboard_interrupt_returns_the_cell | yes | yes | utils.live.poll | #781 | |
+| tests/test_live_poll.py::test_processor_module_is_gone | yes | yes | utils.processor (deleted) | #781 | guards against resurrection |
 | tests/test_dbreader.py::test_missing_column_warns_once | yes | yes | readers.dbreader.Reader._pick_info | #1008 | warn-once per missing header |
 | tests/test_dbreader.py::test_nom_cap_specifics_column_reaches_pages | yes | yes | batch._dbengine._create_pages_dict | #1008 | db value → pages |
 | tests/test_dbreader.py::test_simple_db_engine_skip_file_search_excel_reader | yes | yes | batch._dbengine.simple_db_engine / find_files | #1017 | skip_file_search frames one row per cell |

HISTORY.md

@@ -18,6 +18,13 @@
   nominal capacity, cycle mode and cell name. Works on cells loaded from a
   cellpy-file. Returns `True` when the frames changed. (#164)
 
+* `cellpy.utils.live.poll(cell_or_path, interval, on_update=, until=,
+  max_polls=, timeout=, stop_when_complete=True)`: follow a running test by
+  calling `c.update()` on an interval; the run is recorded on
+  `c.poll_status`. The unused `cellpy.utils.processor` module is removed
+  (its thread-pool fan-out lives in `batch.runner`'s `executor="threads"`).
+  (#781)
+
 ## [2.1.5.post6] - 2026-09-25
 
 * `summary_collector(...).plot()` keeps a lone charge or discharge series

cellpy/readers/cellreader.py

@@ -365,6 +365,9 @@ def __init__(
         #: Incremental-load position for ``update()`` (#164); derived from
         #: the raw frame when None, so it is never persisted.
         self._load_marker = None
+        #: Set by ``update()`` from the loader's ``IncrementalChunk.complete``
+        #: flag; ``live.poll(stop_when_complete=True)`` reads it.
+        self.source_complete = False
         self.debug = debug
         logging.debug("created CellpyCell instance")
 
@@ -2037,6 +2040,7 @@ def _update_incremental(self, fid, loader) -> bool:
         source = fid.full_name if not getattr(fid, "is_db", False) else fid.name
         chunk = loader.load_since(source, marker)
         self._load_marker = chunk.marker
+        self.source_complete = bool(getattr(chunk, "complete", False))
         if chunk.new_raw is None or chunk.new_raw.height == 0:
             self._refresh_fid(fid)
             return False

cellpy/utils/live.py

@@ -1,10 +1,160 @@
-"""Routines for streaming cell data"""
-
-import logging
-import os
-import warnings
-
-logging.captureWarnings(True)
-
-if __name__ == "__main__":
-    warnings.warn("to be implemented")
+"""Follow a running test: poll a cell's raw source and refresh it (#781).
+
+The loop is a thin layer over ``CellpyCell.update()`` (#164): each tick asks the
+cell to re-check its raw source and append what is new; ``on_update`` runs
+after every tick that changed the frames. Nothing here talks to loaders or
+cellpy-core directly.
+
+Examples:
+    ```python
+    from cellpy.utils import live
+
+    def show(c):
+        print(c.data.summary.tail(1))
+
+    c = live.poll("running_test.csv", interval=60, on_update=show, max_polls=10,
+                  instrument="neware_txt", model="UIO")
+    ```
+
+    Stop on your own condition instead of a tick budget:
+
+    ```python
+    c = live.poll(c, interval=30, until=lambda cell: cell.get_cycle_numbers()[-1] >= 50)
+    ```
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Callable, Optional
+
+import cellpy
+from cellpy.readers.cellreader import CellpyCell
+
+logging.captureWarnings(True)
+
+
+class PollStatus:
+    """Outcome of one ``poll`` run (also exposed as ``cell.poll_status``).
+
+    Attributes:
+        polls: ticks run (each tick is one ``update()`` call).
+        updates: ticks where the frames changed.
+        stopped_by: ``"complete"``, ``"until"``, ``"max_polls"``,
+            ``"timeout"``, ``"interrupted"``, or ``"error"``.
+        error: the exception when ``stopped_by == "error"``.
+    """
+
+    def __init__(self):
+        self.polls = 0
+        self.updates = 0
+        self.stopped_by: Optional[str] = None
+        self.error: Optional[BaseException] = None
+
+    def __repr__(self):
+        return f"PollStatus(polls={self.polls}, updates={self.updates}, stopped_by={self.stopped_by!r})"
+
+
+def poll(
+    cell_or_path,
+    interval: float = 30.0,
+    on_update: Optional[Callable] = None,
+    stop_when_complete: bool = True,
+    until: Optional[Callable] = None,
+    max_polls: Optional[int] = None,
+    timeout: Optional[float] = None,
+    raise_errors: bool = False,
+    sleep: Callable[[float], None] = time.sleep,
+    **get_kwargs,
+):
+    """Refresh a cell from its raw source on an interval until a stop condition.
+
+    Args:
+        cell_or_path: a ``CellpyCell`` or a path handed to ``cellpy.get``
+            (with ``**get_kwargs``, e.g. ``instrument=``, ``model=``,
+            ``mass=``). When a path is given, the initial load counts as the
+            first update and ``on_update`` fires for it.
+        interval: seconds to wait between ticks.
+        on_update: ``on_update(cell)``; called after every tick that changed
+            the frames (and after the initial load from a path).
+        stop_when_complete: stop when the loader reported the test has ended
+            (``cell.source_complete``; only loaders that can tell set it).
+        until: ``until(cell) -> bool``; checked after every tick, stops when
+            true.
+        max_polls: stop after this many ticks (``None`` = no limit).
+        timeout: stop after this many seconds (``None`` = no limit).
+        raise_errors: re-raise exceptions from ``update``/``on_update``
+            instead of stopping the loop with ``stopped_by="error"``.
+        sleep: the wait function (injectable for tests / event loops).
+        **get_kwargs: forwarded to ``cellpy.get`` when a path is given, and
+            (loader-related keys) to ``cell.update`` on every tick.
+
+    Returns:
+        The cell, with the run recorded on ``cell.poll_status``.
+
+    ``KeyboardInterrupt`` stops the loop cleanly (``stopped_by="interrupted"``).
+    With no stop condition at all the loop runs until interrupted.
+    """
+    status = PollStatus()
+    update_kwargs = {k: v for k, v in get_kwargs.items() if k in _UPDATE_KWARGS}
+
+    if isinstance(cell_or_path, CellpyCell):
+        cell = cell_or_path
+    else:
+        cell = cellpy.get(cell_or_path, **get_kwargs)
+        status.updates += 1
+        _fire(on_update, cell, status, raise_errors)
+
+    cell.poll_status = status
+    started = time.monotonic()
+
+    try:
+        while status.stopped_by is None:
+            if _should_stop(cell, status, stop_when_complete, until, max_polls, timeout, started):
+                break
+            sleep(interval)
+            status.polls += 1
+            try:
+                changed = cell.update(**update_kwargs)
+            except Exception as exc:  # noqa: BLE001 - reported on the status
+                if raise_errors:
+                    raise
+                logging.error(f"poll: update failed ({exc})")
+                status.stopped_by, status.error = "error", exc
+                break
+            if changed:
+                status.updates += 1
+                _fire(on_update, cell, status, raise_errors)
+    except KeyboardInterrupt:
+        status.stopped_by = "interrupted"
+    logging.info(f"poll: finished {status}")
+    return cell
+
+
+#: ``cellpy.get`` kwargs that ``update()`` must see too (loader recreation).
+_UPDATE_KWARGS = ("model",)
+
+
+def _fire(on_update, cell, status, raise_errors):
+    if on_update is None:
+        return
+    try:
+        on_update(cell)
+    except Exception as exc:  # noqa: BLE001 - reported on the status
+        if raise_errors:
+            raise
+        logging.error(f"poll: on_update failed ({exc})")
+        status.stopped_by, status.error = "error", exc
+
+
+def _should_stop(cell, status, stop_when_complete, until, max_polls, timeout, started) -> bool:
+    if stop_when_complete and getattr(cell, "source_complete", False):
+        status.stopped_by = "complete"
+    elif until is not None and until(cell):
+        status.stopped_by = "until"
+    elif max_polls is not None and status.polls >= max_polls:
+        status.stopped_by = "max_polls"
+    elif timeout is not None and time.monotonic() - started >= timeout:
+        status.stopped_by = "timeout"
+    return status.stopped_by is not None

docs/agents/index.md

@@ -178,6 +178,12 @@ Useful methods on `CellpyCell` (non-exhaustive):
   mass / area / nominal capacity / cycle mode. Works on a cell loaded from a
   `.cellpy` file too (the raw path is stored in it); pass loader kwargs
   such as `model="UIO"` when the instrument needs them.
+  To follow a running test on an interval use
+  `from cellpy.utils import live` then
+  `live.poll(c_or_path, interval=60, on_update=callback, max_polls=..., until=...)`;
+  it calls `update()` each tick, runs `on_update(c)` when frames changed,
+  and records the run on `c.poll_status`. Pass `sleep=` to drive it from
+  your own scheduler. `Ctrl-C` stops it cleanly and returns the cell.
 - `save` / `to_csv` / Excel helpers — persist for the user's workflow
 
 Deeper shape docs: [Data structure](../fundamentals/data_structure.md).

docs/contributing/developers_guide/dev_cellpy_folder_structure.md

@@ -115,8 +115,7 @@ The `cellpy` source code is structured as follows:
     ├── 🐍 ica.py
     ├── 🐍 live.py
     ├── 🐍 ocv_rlx.py
-    ├── 🐍 plotutils.py
-    └── 🐍 processor.py
+    └── 🐍 plotutils.py
     ...
 ```