From 49ce184432b6c0ec075b089038bda3748eee9de6 Mon Sep 17 00:00:00 2001 From: jepegit Date: Sat, 26 Sep 2026 00:46:38 +0200 Subject: [PATCH] Add live.poll() and remove the unused utils.processor module (#781) poll() refreshes a cell through CellpyCell.update() on an interval with on_update / until / max_polls / timeout / stop_when_complete stop conditions and an injectable sleep; the run is recorded on cell.poll_status. processor.py's thread-pool sketch is covered by batch.runner's threads executor. Co-authored-by: Cursor --- .../03-solved-issues/issue781_original.md | 15 ++ .issueflows/03-solved-issues/issue781_plan.md | 15 ++ .../03-solved-issues/issue781_status.md | 21 +++ .../incremental-load-protocol.md | 24 ++- .../04-designs-and-guides/test-registry.md | 8 + HISTORY.md | 7 + cellpy/readers/cellreader.py | 4 + cellpy/utils/live.py | 170 ++++++++++++++++-- cellpy/utils/processor.py | 57 ------ docs/agents/index.md | 6 + .../dev_cellpy_folder_structure.md | 3 +- tests/test_live_poll.py | 137 ++++++++++++++ 12 files changed, 393 insertions(+), 74 deletions(-) create mode 100644 .issueflows/03-solved-issues/issue781_original.md create mode 100644 .issueflows/03-solved-issues/issue781_plan.md create mode 100644 .issueflows/03-solved-issues/issue781_status.md delete mode 100644 cellpy/utils/processor.py create mode 100644 tests/test_live_poll.py diff --git a/.issueflows/03-solved-issues/issue781_original.md b/.issueflows/03-solved-issues/issue781_original.md new file mode 100644 index 000000000..e82c33117 --- /dev/null +++ b/.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. diff --git a/.issueflows/03-solved-issues/issue781_plan.md b/.issueflows/03-solved-issues/issue781_plan.md new file mode 100644 index 000000000..e93d71065 --- /dev/null +++ b/.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. diff --git a/.issueflows/03-solved-issues/issue781_status.md b/.issueflows/03-solved-issues/issue781_status.md new file mode 100644 index 000000000..2c9c56878 --- /dev/null +++ b/.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. diff --git a/.issueflows/04-designs-and-guides/incremental-load-protocol.md b/.issueflows/04-designs-and-guides/incremental-load-protocol.md index d18258648..633326155 100644 --- a/.issueflows/04-designs-and-guides/incremental-load-protocol.md +++ b/.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). diff --git a/.issueflows/04-designs-and-guides/test-registry.md b/.issueflows/04-designs-and-guides/test-registry.md index 9f5fdcd9e..6969c6395 100644 --- a/.issueflows/04-designs-and-guides/test-registry.md +++ b/.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 | diff --git a/HISTORY.md b/HISTORY.md index 05f5dc206..9bf9d665c 100644 --- a/HISTORY.md +++ b/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 diff --git a/cellpy/readers/cellreader.py b/cellpy/readers/cellreader.py index 8a06c4fe4..e44fb721a 100644 --- a/cellpy/readers/cellreader.py +++ b/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 diff --git a/cellpy/utils/live.py b/cellpy/utils/live.py index ba3733896..d56e9245f 100644 --- a/cellpy/utils/live.py +++ b/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 diff --git a/cellpy/utils/processor.py b/cellpy/utils/processor.py deleted file mode 100644 index 1b357d052..000000000 --- a/cellpy/utils/processor.py +++ /dev/null @@ -1,57 +0,0 @@ -import cellpy -import multiprocessing -import concurrent.futures -import time -import pathlib -import pandas as pd - -ASYNC_MODE = "threading" - -if ASYNC_MODE == "threading": - PoolExecutor = concurrent.futures.ThreadPoolExecutor -else: - # cellpy.CellpyCell object is not serializable so returning it - # within a process-pool will crash. - # TODO: make it serializable - PoolExecutor = concurrent.futures.ProcessPoolExecutor - - -def func(filename): - # time.sleep(1) - print(f"{filename}: {pathlib.Path(filename).is_file()}") - c = cellpy.get(filename) - return c - - -def _main(): - print(" starting ".center(80, "-")) - max_number_processes = multiprocessing.cpu_count() - print(f"{max_number_processes=}") - f1 = r"C:\scripting\cellpy\testdata\batch_project\data\raw\20230221_CLP001_1_02_cc_01.res" - f2 = r"C:\scripting\cellpy\testdata\batch_project\data\raw\20230221_CLP001_1_03_cc_01.res" - f3 = r"C:\scripting\cellpy\testdata\batch_project\data\raw\20230221_CLP001_2_01_cc_01.res" - f4 = r"C:\scripting\cellpy\testdata\batch_project\data\raw\20230221_CLP001_2_04_cc_01.res" - params = [ - dict(filename=f1), - dict(filename=f2), - dict(filename=f3), - dict(filename=f4), - ] - t0 = time.time() - with PoolExecutor() as executor: - pool = [executor.submit(func, **param) for param in params] - - for i in concurrent.futures.as_completed(pool): - c = i.result(timeout=2000) - dt_p = time.time() - t0 - - t0 = time.time() - for param in params: - c = func(**param) - dt_s = time.time() - t0 - print(f"Parallel processing took {dt_p} seconds") - print(f"Sequential processing took {dt_s} seconds") - - -if __name__ == "__main__": - _main() diff --git a/docs/agents/index.md b/docs/agents/index.md index ffffd19eb..b56f59ba5 100644 --- a/docs/agents/index.md +++ b/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). diff --git a/docs/contributing/developers_guide/dev_cellpy_folder_structure.md b/docs/contributing/developers_guide/dev_cellpy_folder_structure.md index 78a1032c3..8aa2f7a69 100644 --- a/docs/contributing/developers_guide/dev_cellpy_folder_structure.md +++ b/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 ... ``` diff --git a/tests/test_live_poll.py b/tests/test_live_poll.py new file mode 100644 index 000000000..ce751b39f --- /dev/null +++ b/tests/test_live_poll.py @@ -0,0 +1,137 @@ +"""``cellpy.utils.live.poll`` (#781): the poll loop over ``CellpyCell.update()``.""" + +from __future__ import annotations + +import shutil + +import pytest + +import cellpy +from cellpy.utils import live +from tests.incremental_support import ( + NEWARE_KWARGS, + NEWARE_UIO, + assert_cell_frames_equal, + truncate_text_file, +) + +pytestmark = pytest.mark.essential + +HEAD_ROWS = 6000 + + +class FakeClock: + """Collects sleep calls; ``grow_at`` maps tick index → rows to write.""" + + def __init__(self, path, grow_at): + self.path = path + self.grow_at = grow_at + self.sleeps = [] + + def __call__(self, seconds): + self.sleeps.append(seconds) + tick = len(self.sleeps) + if tick in self.grow_at: + rows = self.grow_at[tick] + if rows is None: + shutil.copyfile(NEWARE_UIO, self.path) + else: + truncate_text_file(NEWARE_UIO, self.path, rows) + + +@pytest.fixture +def live_file(tmp_path): + return truncate_text_file(NEWARE_UIO, tmp_path / "live.csv", HEAD_ROWS) + + +def test_poll_updates_on_growth_and_stops_at_max_polls(live_file): + full = cellpy.get(NEWARE_UIO, testing=True, **NEWARE_KWARGS) + c = cellpy.get(live_file, testing=True, **NEWARE_KWARGS) + seen = [] + clock = FakeClock(live_file, {2: 8800, 4: None}) + out = live.poll(c, interval=7, on_update=lambda cell: seen.append(len(cell.data.raw)), max_polls=5, sleep=clock) + assert out is c + assert clock.sleeps == [7] * 5 + assert seen == [8800, len(full.data.raw)] + assert c.poll_status.polls == 5 + assert c.poll_status.updates == 2 + assert c.poll_status.stopped_by == "max_polls" + assert_cell_frames_equal(c, full) + + +def test_poll_from_path_fires_on_update_for_the_initial_load(live_file): + seen = [] + clock = FakeClock(live_file, {}) + c = live.poll( + live_file, + interval=1, + on_update=lambda cell: seen.append(len(cell.data.raw)), + max_polls=1, + sleep=clock, + testing=True, + **NEWARE_KWARGS, + ) + assert seen == [HEAD_ROWS] + assert c.poll_status.updates == 1 + assert c.poll_status.polls == 1 + + +def test_poll_stops_on_until(live_file): + c = cellpy.get(live_file, testing=True, **NEWARE_KWARGS) + clock = FakeClock(live_file, {1: None}) + live.poll(c, interval=1, until=lambda cell: len(cell.data.raw) > HEAD_ROWS, sleep=clock) + assert c.poll_status.stopped_by == "until" + assert c.poll_status.polls == 1 + + +def test_poll_stops_when_source_complete(live_file): + c = cellpy.get(live_file, testing=True, **NEWARE_KWARGS) + c.source_complete = True + clock = FakeClock(live_file, {}) + live.poll(c, interval=1, sleep=clock) + assert c.poll_status.stopped_by == "complete" + assert clock.sleeps == [] + + +def test_poll_stops_on_timeout(live_file, monkeypatch): + c = cellpy.get(live_file, testing=True, **NEWARE_KWARGS) + now = [0.0] + monkeypatch.setattr(live.time, "monotonic", lambda: now[0]) + + def clock(seconds): + now[0] += seconds + + live.poll(c, interval=10, timeout=25, sleep=clock) + assert c.poll_status.stopped_by == "timeout" + assert c.poll_status.polls == 3 + + +def test_poll_records_update_errors_unless_raise_errors(live_file, monkeypatch): + c = cellpy.get(live_file, testing=True, **NEWARE_KWARGS) + + def boom(**kwargs): + raise RuntimeError("tester unplugged") + + monkeypatch.setattr(c, "update", boom) + live.poll(c, interval=1, max_polls=3, sleep=lambda s: None) + assert c.poll_status.stopped_by == "error" + assert isinstance(c.poll_status.error, RuntimeError) + assert c.poll_status.polls == 1 + with pytest.raises(RuntimeError): + live.poll(c, interval=1, max_polls=3, sleep=lambda s: None, raise_errors=True) + + +def test_poll_keyboard_interrupt_returns_the_cell(live_file): + c = cellpy.get(live_file, testing=True, **NEWARE_KWARGS) + + def interrupt(seconds): + raise KeyboardInterrupt + + out = live.poll(c, interval=1, sleep=interrupt) + assert out is c + assert c.poll_status.stopped_by == "interrupted" + + +def test_processor_module_is_gone(): + with pytest.raises(ImportError): + import cellpy.utils.processor # noqa: F401