{"name":"nbatch","display_name":"nbatch","visibility":"public","icon":"","categories":["Utilities"],"schema_version":"0.2.1","on_activate":null,"on_deactivate":null,"contributions":{"commands":[],"readers":null,"writers":null,"widgets":null,"sample_data":null,"themes":null,"menus":{},"submenus":null,"keybindings":null,"configuration":[]},"package_metadata":{"metadata_version":"2.4","name":"nbatch","version":"0.0.4","dynamic":["license-file"],"platform":null,"supported_platform":null,"summary":"Flexible batch processing utilities","description":"# nbatch\n\n[![License BSD-3](https://img.shields.io/pypi/l/nbatch.svg?color=green)](https://github.com/ndev-kit/nbatch/raw/main/LICENSE)\n[![PyPI](https://img.shields.io/pypi/v/nbatch.svg?color=green)](https://pypi.org/project/nbatch)\n[![Python Version](https://img.shields.io/pypi/pyversions/nbatch.svg?color=green)](https://python.org)\n[![tests](https://github.com/ndev-kit/nbatch/workflows/tests/badge.svg)](https://github.com/ndev-kit/nbatch/actions)\n[![codecov](https://codecov.io/gh/ndev-kit/nbatch/branch/main/graph/badge.svg)](https://codecov.io/gh/ndev-kit/nbatch)\n[![napari hub](https://img.shields.io/endpoint?url=https://api.napari-hub.org/shields/nbatch)](https://napari-hub.org/plugins/nbatch)\n[![npe2](https://img.shields.io/badge/plugin-npe2-blue?link=https://napari.org/stable/plugins/index.html)](https://napari.org/stable/plugins/index.html)\n[![Copier](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/copier-org/copier/master/img/badge/badge-grayscale-inverted-border-purple.json)](https://github.com/copier-org/copier)\n\n**Lightweight batch processing utilities for the ndev-kit ecosystem.**\n\nnbatch provides a foundation for batch processing operations. It's designed to work seamlessly with napari plugins but has **no napari or Qt dependencies**.\n\n## Features\n\n- **`@batch` decorator** - Transform single-item functions into batch-capable functions\n- **`BatchContext`** - Track progress through batch operations\n- **`BatchRunner`** - Orchestrate batch operations with threading, progress callbacks, and cancellation\n- **`discover_files()`** - Flexible file discovery with natural sorting (like file explorers)\n- **`batch_logger`** - Scoped logging for batch operations with headers/footers\n- **Minimal dependencies** - Only requires natsort for natural file ordering\n- **Optional napari integration** - Uses napari's threading when available, falls back to standard threads\n\n## Installation\n\n```bash\npip install nbatch\n```\n\nFor development:\n\n```bash\npip install -e . --group dev\n```\n\n## Quick Start\n\n### Basic Batch Processing\n\nThe `@batch` decorator transforms a function that processes a single item into one that handles both single items and batches:\n\n```python\nfrom pathlib import Path\nfrom nbatch import batch\n\n@batch\ndef process_image(path: Path) -> str:\n    # Your processing logic here\n    return path.stem.upper()\n\n# Single item - returns result directly\nresult = process_image(Path(\"image.tif\"))\n# Returns: \"IMAGE\"\n\n# List of items - returns generator\nresults = process_image([Path(\"a.tif\"), Path(\"b.tif\")])\nlist(results)\n# Returns: [\"A\", \"B\"]\n\n# Directory - discovers files and returns generator\nresults = process_image(Path(\"/data/images\"))\n# Processes all files in directory\n```\n\n### Progress Tracking\n\nUse `with_context=True` to get progress information:\n\n```python\n@batch(with_context=True)\ndef process_image(path: Path) -> str:\n    return path.stem\n\nfor result, ctx in process_image(files):\n    print(f\"{ctx.progress:.0%} complete: {result}\")\n    # 10% complete: image1\n    # 20% complete: image2\n    # ...\n```\n\nThe `BatchContext` provides:\n\n- `ctx.index` - Zero-based index of current item\n- `ctx.total` - Total number of items\n- `ctx.item` - The current item being processed\n- `ctx.progress` - Progress as fraction (0.0 to 1.0)\n- `ctx.is_first` / `ctx.is_last` - Boolean flags\n\n### Error Handling\n\nControl how errors are handled with `on_error`:\n\n```python\n# 'raise' (default) - Re-raise exceptions immediately\n@batch(on_error='raise')\ndef strict_process(path): ...\n\n# 'continue' - Log error and yield None for failed items\n@batch(on_error='continue')\ndef lenient_process(path): ...\n# Results: [\"good\", None, \"ok\"]\n\n# 'skip' - Log error and skip failed items entirely\n@batch(on_error='skip')\ndef skip_errors(path): ...\n# Results: [\"good\", \"ok\"]\n```\n\n### File Discovery\n\nControl which files are processed:\n\n```python\n# Custom glob patterns\n@batch(patterns='*.tif')\ndef process_tiffs(path): ...\n\n# Multiple patterns\n@batch(patterns=['*.tif', '*.tiff', '*.png'])\ndef process_images(path): ...\n\n# Non-recursive (top-level only)\n@batch(recursive=False)\ndef process_top_level(path): ...\n```\n\nOr use `discover_files()` directly:\n\n```python\nfrom nbatch import discover_files\n\n# From directory with patterns\nfiles = discover_files(\"/data/images\", patterns=[\"*.tif\", \"*.png\"])\n\n# From explicit list\nfiles = discover_files([path1, path2, path3])\n```\n\n### Logging\n\nUse `batch_logger` for structured logging. By default, it outputs to the console (stderr). Optionally log to a file:\n\n```python\nfrom nbatch import batch, batch_logger\n\n@batch(with_context=True)\ndef process(path):\n    return path.stem\n\n# Console only (default)\nwith batch_logger() as log:\n    for result, ctx in process(files):\n        log(ctx, f\"Processed: {result}\")\n\n# With file logging (appends by default)\nwith batch_logger(log_file=\"output/process.log\", header={\"Files\": 100}) as log:\n    for result, ctx in process(files):\n        log(ctx, f\"Processed: {result}\")\n        # Or use log.info(), log.warning(), log.error()\n\n# File only (no console output)\nwith batch_logger(log_file=\"output/quiet.log\", console=False) as log:\n    for result, ctx in process(files):\n        log(ctx, f\"Processed: {result}\")\n```\n\nLog file output:\n```\n============================================================\nBatch processing started at 2025-01-29 10:30:00\n------------------------------------------------------------\nFiles: 100\n============================================================\n2025-01-29 10:30:01 - INFO - [1/100] image1.tif - Processed: image1\n2025-01-29 10:30:02 - INFO - [2/100] image2.tif - Processed: image2\n...\n============================================================\nBatch processing completed at 2025-01-29 10:35:00\n============================================================\n```\n\n## Integration with napari\n\n### Using BatchRunner (Recommended)\n\n`BatchRunner` provides clean orchestration for widgets with threading, progress callbacks, and cancellation:\n\n```python\nfrom nbatch import batch, BatchRunner\n\n# Define your processing function (pure, testable)\n@batch(on_error='continue')\ndef process_image(path, model, output_dir):\n    result = model.predict(load_image(path))\n    save_result(result, output_dir / path.name)\n    return result\n\n# In your widget class\nclass MyWidget:\n    def __init__(self, viewer):\n        self._viewer = viewer\n        \n        # Create runner once - reusable for all batches\n        self.runner = BatchRunner(\n            on_start=self._on_batch_start,\n            on_item_complete=self._on_item_complete,\n            on_complete=self._on_batch_complete,\n            on_error=self._on_item_error,\n            on_cancel=self._on_cancelled,\n        )\n        \n        self._run_button.clicked.connect(self.run_batch)\n        self._cancel_button.clicked.connect(self.runner.cancel)\n    \n    def _on_batch_start(self, total):\n        \"\"\"Called when batch starts with total item count.\"\"\"\n        self._progress_bar.setValue(0)\n        self._progress_bar.setMaximum(total)\n    \n    def _on_item_complete(self, result, ctx):\n        \"\"\"Called after each item completes.\"\"\"\n        self._progress_bar.setValue(ctx.index + 1)\n        # Optionally add result to viewer\n        if result is not None:\n            self._viewer.add_image(result, name=f\"Result {ctx.index}\")\n    \n    def _on_batch_complete(self):\n        errors = self.runner.error_count\n        if errors > 0:\n            self._progress_bar.label = f\"Done with {errors} errors\"\n        else:\n            self._progress_bar.label = \"Complete!\"\n    \n    def _on_item_error(self, ctx, exception):\n        self._progress_bar.label = f\"Error on {ctx.item.name}\"\n    \n    def _on_cancelled(self):\n        self._progress_bar.label = \"Cancelled\"\n    \n    def run_batch(self):\n        \"\"\"Triggered by 'Run' button - just one line!\"\"\"\n        self.runner.run(\n            process_image,\n            self.files,\n            model=self.model,\n            output_dir=self.output_dir,\n            log_file=self.output_dir / \"batch.log\",\n        )\n```\n\n### Using @thread_worker directly\n\nFor more control, use napari's `@thread_worker` with the `@batch` decorator:\n\n```python\nfrom napari.qt.threading import thread_worker\nfrom nbatch import batch, batch_logger\n\n@batch(with_context=True, on_error='continue')\ndef process_image(path, model, output_dir):\n    # Your processing logic\n    result = model.predict(load_image(path))\n    save_result(result, output_dir / path.name)\n    return result\n\n# In your widget\ndef run_batch(self):\n    @thread_worker\n    def _run():\n        with batch_logger(log_file=self.output_dir / 'log.txt') as log:\n            for result, ctx in process_image(\n                self.input_dir,\n                model=self.model,\n                output_dir=self.output_dir,\n            ):\n                log(ctx, f\"Processed: {ctx.item.name}\")\n                yield ctx  # Enables progress updates\n    \n    worker = _run()\n    worker.yielded.connect(\n        lambda ctx: self.progress_bar.setValue(int(ctx.progress * 100))\n    )\n    worker.start()\n```\n\n## API Reference\n\n### `@batch` Decorator\n\n```python\n@batch(\n    on_error: Literal['raise', 'continue', 'skip'] = 'raise',\n    with_context: bool = False,\n    patterns: str | Sequence[str] = '*',\n    recursive: bool = False,\n)\n```\n\n### `BatchContext`\n\n```python\n@dataclass(frozen=True)\nclass BatchContext:\n    index: int      # Zero-based index\n    total: int      # Total items\n    item: Any       # Current item\n    \n    @property\n    def progress(self) -> float: ...  # (index + 1) / total\n    @property\n    def is_first(self) -> bool: ...   # index == 0\n    @property\n    def is_last(self) -> bool: ...    # index == total - 1\n```\n\n### `discover_files()`\n\n```python\ndef discover_files(\n    source: str | Path | Iterable[str | Path],\n    patterns: str | Sequence[str] = '*',\n    recursive: bool = False,\n) -> list[Path]: ...\n```\n\n### `batch_logger`\n\n```python\n@contextmanager\ndef batch_logger(\n    log_file: str | Path | None = None,  # Optional file path\n    header: Mapping[str, object] | None = None,  # Metadata to write at start\n    level: int = logging.INFO,\n    console: bool = True,  # Output to stderr\n    file_mode: Literal['w', 'a'] = 'a',  # Append by default\n) -> Generator[BatchLogger, None, None]: ...\n```\n\n### `BatchRunner`\n\n```python\nclass BatchRunner:\n    def __init__(\n        self,\n        on_start: Callable[[int], None] | None = None,\n        on_item_complete: Callable[[Any, BatchContext], None] | None = None,\n        on_complete: Callable[[], None] | None = None,\n        on_error: Callable[[BatchContext, Exception], None] | None = None,\n        on_cancel: Callable[[], None] | None = None,\n    ): ...\n    \n    def run(\n        self,\n        func: Callable,\n        items: Any,\n        *args,\n        threaded: bool = True,\n        log_file: str | Path | None = None,\n        log_header: Mapping[str, object] | None = None,\n        patterns: str | Sequence[str] = '*',\n        recursive: bool = False,\n        **kwargs,  # Passed to func!\n    ) -> None: ...\n    \n    def cancel(self) -> None: ...\n    \n    @property\n    def is_running(self) -> bool: ...\n    \n    @property\n    def was_cancelled(self) -> bool: ...\n    \n    @property\n    def error_count(self) -> int: ...  # Errors in current/last batch\n```\n\n## Contributing\n\nContributions are welcome! Please ensure tests pass before submitting a pull request:\n\n```bash\npytest --cov=src/nbatch\n```\n\n## License\n\nDistributed under the terms of the [BSD-3](http://opensource.org/licenses/BSD-3-Clause) license.\n\n## Part of ndev-kit\n\nnbatch is part of the [ndev-kit](https://github.com/ndev-kit) ecosystem for no-code bioimage analysis in napari.\n","description_content_type":"text/markdown","keywords":null,"home_page":null,"download_url":null,"author":"Tim Monko","author_email":"timmonko@gmail.com","maintainer":null,"maintainer_email":null,"license":null,"classifier":["Development Status :: 2 - Pre-Alpha","Framework :: napari","Intended Audience :: Developers","Operating System :: OS Independent","Programming Language :: Python","Programming Language :: Python :: 3","Programming Language :: Python :: 3 :: Only","Programming Language :: Python :: 3.10","Programming Language :: Python :: 3.11","Programming Language :: Python :: 3.12","Programming Language :: Python :: 3.13","Topic :: Scientific/Engineering :: Image Processing"],"requires_dist":["natsort","napari; extra == \"napari\"","pyqt6; extra == \"qtpy-backend\"","nbatch[napari]; extra == \"all\"","nbatch[qtpy-backend]; extra == \"all\""],"requires_python":">=3.10","requires_external":null,"project_url":["Bug Tracker, https://github.com/ndev-kit/nbatch/issues","Documentation, https://github.com/ndev-kit/nbatch#README.md","Source Code, https://github.com/ndev-kit/nbatch","User Support, https://github.com/ndev-kit/nbatch/issues"],"provides_extra":["napari","qtpy-backend","all"],"provides_dist":null,"obsoletes_dist":null},"npe1_shim":false}