Metadata-Version: 2.4
Name: scs-architecture-handlers
Version: 0.2.9
Summary: ArchitectureHandlers for all parts of the Maverick Architecture
Author: SCS
License: MIT
Project-URL: Homepage, https://gitlab.ub.uni-bielefeld.de/scs/enrico/modules/architecture-handlers
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: numpy>=1.24
Requires-Dist: requests>=2.31
Requires-Dist: Pillow>=9.5
Requires-Dist: realtimestt-multiclient>=0.2.1
Requires-Dist: ollama
Requires-Dist: opencv-python
Requires-Dist: pytest
Requires-Dist: pytest-asyncio>=1.4.0
Requires-Dist: aiortc>=1.5.0
Requires-Dist: httpx>=0.24.0
Requires-Dist: av>=10.0.0
Requires-Dist: librosa>=0.10.0

# Architecture Handlers

A minimal, extensible Python library providing a base class (`ArchitectureHandler`) for building handler components that accept inputs, optionally perform network requests, and publish results via thread-safe queues. Ships with concrete handlers for AI/agent tasks: speech-to-text, text-to-speech, face recognition, gesture generation, LLM agent interaction, and Ollama chat.

**Package:** `scs-architecture-handlers` · **Python:** >= 3.10 · **License:** MIT

---

## Table of Contents

- [Features](#features)
- [Installation](#installation)
- [Handlers Overview](#handlers-overview)
- [Design Decisions](#design-decisions)
- [Usage](#usage)
  - [Direct synchronous generation](#1-direct-synchronous-generation)
  - [Threaded pipeline](#2-threaded-pipeline)
  - [Async wrappers](#3-async-wrappers)
  - [Callbacks & non-blocking execution](#4-callbacks--non-blocking-execution)
- [Extension Points](#extension-points)
- [Configuration Reference](#configuration-reference)
- [Stats & Monitoring](#stats--monitoring)
- [Development](#development)
  - [Setup](#setup)
  - [Running Tests](#running-tests)
  - [System Dependencies](#system-dependencies)
  - [CI/CD & Releasing to PyPI](#cicd--releasing-to-pypi)
- [License](#license)

---

## Features

- **Threaded processing loop** — consumes items from an input queue, processes them, and writes results to a result queue.
- **Pluggable core logic** — override `generate_results()` or supply `generate_results_callback` in config.
- **Template methods** — `validate_item`, `preprocess_item`, `postprocess_result` for easy adaptation.
- **Optional type checking** — `expected_type` catches mismatched inputs at the boundary.
- **Dropping policy** — bounded queues with configurable capacity; drops are counted and logged.
- **Runtime instrumentation** — timestamps and counters (processed/dropped/errors/network errors) via `stats()`.
- **Async wrappers** — all sync methods have `async_*` counterparts (powered by `asyncio.to_thread`).
- **Hook methods** — 9 instrumentation hooks (`_hook_before_feed`, `_hook_after_generate`, etc.) for monitoring.
- **Non-blocking callbacks** — dedicated callback worker thread so user callbacks never stall the hot path.
- **Singleton sync core** — async wrappers delegate to the same sync implementation; no code duplication.

---

## Installation

### From PyPI

```bash
pip install scs-architecture-handlers
```

```python
from scs_architecture_handlers.base_handler import ArchitectureHandler
```

### Development setup

```bash
# Clone the repository
git clone <repo-url>
cd architecture-handlers

# Create a virtual environment (Python >= 3.10)
python3.11 -m venv .venv
source .venv/bin/activate

# Install dependencies (using pip)
pip install -r requirements.txt

# Or using uv
uv sync
```

---

## Handlers Overview

| Handler | File | Purpose |
|---------|------|---------|
| `ArchitectureHandler` | `base_handler.py` | Extensible base class for all handlers. |
| `FaceRecognitionHandler` | `face_recognition_handler.py` | Posts camera frames to a `/detect` FastAPI endpoint. Optional webcam auto-capture mode. |
| `GestureGenerationHandler` | `gesture_generation_handler.py` | Sends audio chunks to a `/generate` gesture FastAPI for realtime beat-driven gesture synthesis. |
| `LLMAgentHandler` | `llm_agent_handler.py` | Sends text or person data to an LLM agent FastAPI service. |
| `STTHandler` | `stt_handler.py` | Speech-to-text via `RealtimeSTT.AudioToTextRecorderClient` (WebSocket). |
| `RealtimeSTTHandler` (v2) | `stt_handler_v2.py` | Speech-to-text via WebRTC / `aiortc` with `_AudioTrackProxy` and `_MicrophoneCapturer`. |
| `F5TTSHandler` | `f5tts_handler.py` | Text-to-speech via an F5-TTS FastAPI `/synthesize` endpoint. |
| `MaverickAgentHandler` | `maverick_agent_handler.py` | Orchestrator: TTS + gesture generation + Unity visualizer streaming via WebSocket. |
| `OllamaChatAgent` | `ollama_handler.py` | Stateful chat wrapper around `ollama.Client` with conversation history, save/load, and error types. |

---

## Design Decisions

1. **Single sync core, async wrappers on top.** All async methods delegate to the sync implementation via `asyncio.to_thread`. This means there is exactly one implementation of the processing logic — no duplicated sync/async code paths.

2. **Template Method pattern.** The base class defines the skeleton of the feed → validate → preprocess → generate → postprocess → publish pipeline. Subclasses override only the steps they need (`validate_item`, `preprocess_item`, `generate_results`, `postprocess_result`).

3. **Bounded queues with backpressure.** Both input and result queues have configurable max sizes. When a queue is full, items are silently dropped and `dropped_count` is incremented. This prevents unbounded memory growth under load.

4. **Dedicated async callback thread.** External callbacks (e.g., database writes, network calls) can be scheduled via `_emit_async_callback` to run on a separate worker thread, keeping the main processing path fast and predictable.

5. **Hook-based instrumentation.** All major lifecycle events fire hook methods. Subclasses and monitoring tools can override these hooks without modifying the core pipeline.

6. **Dry-run mode in all network handlers.** Every handler that performs network I/O supports a `dry_run` flag that returns stub results without making HTTP calls. This makes testing and offline development straightforward.

7. **Weak reference finalizer.** A `weakref.finalize` ensures background threads are joined when the handler is garbage-collected, preventing orphaned threads.

---

## Usage

### 1) Direct synchronous generation

Subclass `ArchitectureHandler` and override `generate_results`:

```python
from scs_architecture_handlers.base_handler import ArchitectureHandler


class EchoHandler(ArchitectureHandler):
    def generate_results(self, item, extra):
        return {"echo": item, "meta": extra}


h = EchoHandler(run_as_thread=False)
out = h.generate("hello", tag=1)
print(out)  # {"echo": "hello", "meta": {"tag": 1}}

h.cleanup()
```

### 2) Threaded pipeline

Use the internal worker thread to consume items from the input queue and push processed results to the result queue:

```python
import time
from scs_architecture_handlers.base_handler import ArchitectureHandler


class Doubler(ArchitectureHandler):
    def generate_results(self, item, extra):
        return item * 2


h = Doubler(run_as_thread=True, disable_thread=False)
h.feed(10)
print(h.get_result(timeout=1.0))  # 20
h.cleanup()
```

### 3) Async wrappers

All public sync methods have async counterparts:

```python
import asyncio
from scs_architecture_handlers.base_handler import ArchitectureHandler


class EchoHandler(ArchitectureHandler):
    def generate_results(self, item, extra):
        return {"echo": item}


async def main():
    h = EchoHandler(run_as_thread=False)
    out = await h.async_generate("async-hello")
    print(out)  # {"echo": "async-hello"}
    await h.async_cleanup()


asyncio.run(main())
```

Available async wrappers: `async_feed`, `async_get_result`, `async_generate`, `async_start`, `async_cleanup`.

### 4) Callbacks & non-blocking execution

Schedule external callbacks on a dedicated thread via `_emit_async_callback`:

```python
import time
from scs_architecture_handlers.base_handler import ArchitectureHandler


def on_result(item, extra):
    time.sleep(0.1)
    print("callback got:", item)


class NonBlockingHandler(ArchitectureHandler):
    def generate_results(self, item, extra):
        result = {"value": str(item).upper(), "extra": extra}
        self._emit_async_callback(result, extra)
        return result


h = NonBlockingHandler(
    run_as_thread=False,
    disable_thread=False,
    generate_results_callback=on_result,
    callback_queue_size=8,
)

for i in range(5):
    out = h.generate(f"msg-{i}")
    assert out["value"] == f"MSG-{i}"

time.sleep(0.5)
h.cleanup()
```

Notes:
- If the callback queue fills, new tasks are dropped and `dropped_count` increments.
- For producer-style handlers (e.g., microphone/STT), use `_emit_async_callback` inside the production path.

---

## Extension Points

Override these template methods to adapt the handler:

| Method | Purpose |
| ------ | ------- |
| `validate_item(item, extra) -> bool` | Return `False` to drop invalid items early (shape/content/type). |
| `preprocess_item(item, extra) -> (item, extra)` | Normalize or transform raw input before `generate_results`. |
| `generate_results(item, extra) -> Any` | Core computation (required — must override or supply callback). |
| `postprocess_result(item, extra, result) -> Any` | Final transformation (e.g., map indices to labels). |
| `_prepare_request_payload(item, extra) -> dict` | Build structured payload for network I/O. |
| `_perform_request(payload) -> Any` | Implement actual network request (override this, call via `perform_request`). |

Instrumentation hooks (optional overrides):

| Hook | When it fires |
|------|---------------|
| `_hook_before_feed` / `_hook_after_feed` | Before/after `feed()` |
| `_hook_before_generate` / `_hook_after_generate` | Before/after `generate_results()` |
| `_hook_before_network_call` / `_hook_after_network_call` | Before/after a network request |
| `_hook_after_result` | When a result is retrieved via `get_result()` |
| `_hook_on_error` | When generation raises an exception |
| `_hook_network_error` | When a network call fails |

---

## Configuration Reference

| Key | Default | Description |
|-----|---------|-------------|
| `host` | `None` | Remote host (optional). |
| `port` | `None` | Remote port (optional). |
| `auth` | `None` | Authentication token/credentials (optional). |
| `run_as_thread` | `True` | Start internal worker threads. |
| `disable_thread` | `False` | Force-disable threading even if `run_as_thread` is `True`. |
| `max_queue_size` | `128` | Capacity for the input queue. |
| `result_queue_size` | `max_queue_size` | Capacity for the result queue. |
| `verbose` | `False` | Enable INFO-level logging (default is WARNING). |
| `expected_type` | `None` | Type or tuple for basic validation in `validate_item`. |
| `generate_results_callback` | `None` | External callback alternative to subclassing `generate_results`. |
| `network_timeout` | `None` | Optional timeout for network requests. |
| `client_id_prefix` | `"client"` | Prefix for auto-generated client ID. |
| `callback_queue_size` | `128` | Capacity for the async callback queue (used with `_emit_async_callback`). |

---

## Stats & Monitoring

```python
stats = handler.stats()
print(stats)
# {
#   "client_id": "client-...",
#   "created_at": ...,
#   "last_feed_at": ...,
#   "last_generate_at": ...,
#   "last_result_at": ...,
#   "last_network_call_at": ...,
#   "input_queue_size": 0,
#   "result_queue_size": 0,
#   "run_as_thread": True,
#   "processed_count": 42,
#   "dropped_count": 0,
#   "error_count": 0,
#   "network_error_count": 0,
# }
```

All counters are updated automatically by the default hook implementations. Override hooks for custom monitoring.

---

## Development

### Setup

```bash
python3.11 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

### Running Tests

Tests use `pytest` with `pytest-asyncio` and `pytest-timeout`. They mock external services so no live servers are needed.

```bash
pytest -q
# or
pytest tests/
```

Individual test files:

| Test File | Tests For |
|-----------|-----------|
| `tests/test_base_handler.py` | Core `ArchitectureHandler` (init, feed, generate, threading, hooks, callbacks, cleanup, stats) |
| `tests/test_f5tts_handler.py` | `F5TTSHandler` |
| `tests/test_face_recognition_handler.py` | `FaceRecognitionHandler` |
| `tests/test_gesture_generation_handler.py` | `GestureGenerationHandler` |
| `tests/test_llm_agent_handler.py` | `LLMAgentHandler` |
| `tests/test_maverick_agent_handler.py` | `MaverickAgentHandler` |
| `tests/test_stt_handler.py` | `STTHandler` (v1 / RealtimeSTT) |
| `tests/test_realtime_stt_handler.py` | `STTHandler` (v2 / WebRTC) |

### System Dependencies

On Debian/Ubuntu, you may need:

```bash
sudo apt-get update
sudo apt-get install -y portaudio19-dev
```

### CI/CD & Releasing to PyPI

The project uses GitLab CI (`.gitlab-ci.yml`) with four stages:
1. **test** — runs `pytest -q` on all branches and merge requests.
2. **secret-detection** — GitLab Secret Detection template.
3. **build** — builds sdist+wheel on version tags.
4. **deploy** — publishes to PyPI via Twine on version tags.

To release a new version:

```bash
# 1. Update version in pyproject.toml
# 2. Commit and tag
git tag v0.2.6
git push origin v0.2.6
```

The deploy job verifies the tag matches `pyproject.toml` version automatically.

**Prerequisites:**
- Set a protected, masked CI variable `PYPI_TOKEN` with your PyPI API token in GitLab → Settings → CI/CD → Variables.
- Optionally override `PYPI_REPOSITORY_URL` to point at TestPyPI for testing.

---

## License

MIT
