sientia-dataops-scouter_temporal/CLAUDE.md

# CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

## What this is

Temporal.io worker for industrial time-series data ingestion (`scouter`). Pulls data from MongoDB / PI Web API, runs quality-gate filtering + aggregation, writes to PostgreSQL, caches state in Redis. Heavy dependency on the private `sientia_do` library (aka `sientia-dataops-library`) for base activity classes (Postgres/Redis/Mongo repos), logging, notifications, metrics, retry policy, and connector config builders — read that lib's source when a base method/behavior isn't in this repo.

## Setup

```bash
python3.11 -m venv venv
source ./venv/bin/activate
pip install -r requirements-local.txt   # uses git+ssh to sientia-dataops-library (needs SSH access)
pip install -r requirements-dev.txt     # lint/test/e2e tooling
cp .env.example .env                    # fill in connection details
```

`requirements.txt` uses the plain package name `sientia_do` (resolved via index/private registry in CI); `requirements-local.txt` pins it via `git+ssh://...sientia-dataops-library.git@<tag>` for local dev — keep both in sync when bumping the library version (see `git-requirements-mapping.txt`).

## Commands

```bash
./run_local.sh                          # activate venv, load .env, run worker
python -m scouter.worker.worker          # run worker manually

pytest                                   # unit tests (tests/), Docker-free
pytest tests/activities/test_redis.py    # single file
pytest tests/workflow/test_scouter.py::TestClassName::test_name  # single test
pytest --cov=scouter --cov-report=html   # coverage (or ./run_coverage.sh, opens browser)

# e2e tests are excluded from default `pytest` runs (testpaths=tests) — invoke explicitly:
pytest e2e/ --override-ini testpaths=e2e -m e2e -v
pytest e2e/ --override-ini testpaths=e2e -m e2e -x    # stop on first failure

ruff check .                             # lint
ruff format .                             # format
mypy scouter                             # type check
bandit -r scouter                        # security scan
```

CI quality gate (`.github/workflows/quality-gate.yml`) calls a shared reusable workflow (`Aignosi/github_workflow_templates`) with `requirements_file: requirements-local.txt`; releases (`release.yml`) fire on merged PRs to `main` via another shared workflow. Both are effectively opaque here — the actual lint/test/coverage steps live in the `github_workflow_templates` repo.

## Architecture

Parent/child Temporal workflow pattern, two independent worker task queues sharing the same `CoreScouter` sub-workflow:

```
Scouter (scouter/workflow/scouter.py)              PIWebAPIScouter (scouter/workflow/pi_web_api_scouter.py)
  reads raw_{schedule_name} from MongoDB              queries PI Web API directly (get_tag_values)
  tracks last-processed timestamp in Redis             normalizes all timestamps to the batch max
        │                                                     │
        └───────────────► CoreScouter (scouter/workflow/sub_workflows/core_scouter.py) ◄───────┘
                              1. data_quality_gate   (NULL_VALUES_FILTER / OUT_OF_BOUNDS_FILTER, DISCARD policy)
                              2. aggregate_data      (per-tag: lts/avg/mdn/max/min)
                              3. group_and_hold_data (Redis, TTL = retention_time)
                              4. export_data_to_postgres
                              5. write_metrics       (Prometheus)
                              6. store_data_package  (optional, MongoDB, if debug_data_package=true)
```

- **`Activities`** (`scouter/activities/activities.py`) is a multiple-inheritance god-class combining `Postgres` (from `sientia_do`), `Redis`, `Gates`, `MongoDB`, `API` (this repo's `scouter/activities/*.py`). Each mixin owns its own `__init__`/`close` — the worker calls all of them explicitly. When adding a new activity, decide which mixin it belongs to; don't bypass the split.
- **Temporal sandboxing**: every module doing non-workflow-safe imports (pandas, sientia_do, os, etc.) wraps them in `with workflow.unsafe.imports_passed_through():`. Follow this pattern in any new workflow/activity file — Temporal's sandbox will fail otherwise.
- **Two task queues, one worker process**: `worker.py` builds two `prepare_worker(...)` instances (main workflow `Scouter` vs `PIWebAPIScouter`, both with `CoreScouter` as a registered child) and runs them concurrently via `asyncio.gather`. If either crashes, the whole process exits non-zero (see the `finally` block) — that's deliberate, for k8s restart semantics.
- **Aggregation functions**: `lts` (latest), `avg`, `mdn` (median), `max`, `min` — configured per-tag in `model_tags[tag].aggr_function`.
- **Quality filters**: `NULL_VALUES_FILTER`, `OUT_OF_BOUNDS_FILTER` (`scouter/utils/quality/filters.py`), each with a configurable `policy` (currently `DISCARD`). Register new filters in the `quality_gate_filters` dict in `scouter/activities/gates.py`.
- **Config builders** (`build_postgres_config`, `build_redis_config`, `build_mongodb_config`, `build_api_config`) come from `sientia_do.utils.connectors_config`, not this repo — `scouter/utils/connectors_config.py` only has `build_kafka_config` locally.
- **Metrics** (`scouter/metrics.py`): `APP_UP` gauge, `LABORIOUS_DATA_WRITTEN_COUNT` counter, `TAG_CHANGES_MONITOR` gauge, all labeled by `pod_id`/`model_name`/`workflow_name` (+`tag_name` for the last one). `sonar-project.properties` excludes `scouter/worker/worker.py` from coverage — it's the process entrypoint, exercised by e2e, not unit tests.

## Testing structure

- `tests/` — Docker-free unit tests, mirrors `scouter/` package layout (`tests/activities/`, `tests/workflow/`, `tests/utils/`). Default `pytest` invocation only runs this dir (`testpaths=tests` in `pyproject.toml`).
- `e2e/` — production-faithful tests against real testcontainers (Postgres/Mongo/Redis) + `WorkflowEnvironment.start_local()` + an in-process PI Web API test server (`e2e/pi_web_api_test_server.py`). Only `Logger` and an optional notification-insert spy are mocked; everything else is real `Activities` wiring. Must be invoked explicitly (see Commands) — not picked up by default `pytest`.
  - Scenario catalog lives in `e2e/scenarios.md`, numbered and mapped to `test_scenario_*` functions. Section `## 0` = harness smoke tests (`e2e/test_harness_smoke.py`) that check Docker/container/worker wiring liveness before the real scenarios run.
  - A scenario that fails due to a genuine production defect gets marked `xfail(strict=True)` rather than deleted/skipped silently.
- Markers: `asyncio`, `e2e`, `integration`, `unit` (declared in `pyproject.toml`).

## Root-level scripts

`get_data_pims.py`, `pi_web_api_fetch_data.py`, `init_port_forward.sh` are standalone helper/debug scripts (PI/PIMS data fetch, k8s port-forwarding for local dev against a real cluster) — not part of the worker's runtime path.