Metadata-Version: 2.5
Name: aio-sensorthings
Version: 0.8.0
Summary: A python SensorThingsAPI client and sdk
Project-URL: Homepage, https://gitlab.opencode.de/lgvhh/uda/aio-sensorthings
Author-email: Stefan Schuhart <stefan.schuhart@gv.hamburg.de>, Irene Stemmler <irene.stemmler@gv.hamburg.de>
License: MIT
License-File: AUTHORS.md
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering
Classifier: Topic :: Software Development
Classifier: Topic :: Utilities
Classifier: Typing :: Typed
Requires-Python: <3.15,>=3.13
Requires-Dist: glom<26.0.0,>=25.12.0
Requires-Dist: pydantic-settings<3,>=2.0
Requires-Dist: pydantic<3,>=2.0
Provides-Extra: all
Requires-Dist: aiohttp>=3.9; extra == 'all'
Requires-Dist: aiomqtt>=2.0; extra == 'all'
Requires-Dist: confluent-kafka>=2.3; extra == 'all'
Requires-Dist: opentelemetry-api>=1.20; extra == 'all'
Requires-Dist: redis-om>=1.1; extra == 'all'
Requires-Dist: tenacity>=8.0; extra == 'all'
Provides-Extra: http
Requires-Dist: aiohttp>=3.9; extra == 'http'
Requires-Dist: tenacity>=8.0; extra == 'http'
Provides-Extra: kafka
Requires-Dist: confluent-kafka>=2.3; extra == 'kafka'
Provides-Extra: mqtt
Requires-Dist: aiomqtt>=2.0; extra == 'mqtt'
Requires-Dist: tenacity>=8.0; extra == 'mqtt'
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.20; extra == 'otel'
Provides-Extra: redis
Requires-Dist: redis-om>=1.1; extra == 'redis'
Description-Content-Type: text/markdown

# aio-sensorthings

An asyncio Python library and SDK for OGC SensorThings API (STA) servers such as
[FROST-Server](https://github.com/FraunhoferIOSB/FROST-Server).

## Summary

`aio-sensorthings` copies data from one SensorThings server to another and keeps it
up to date in near real time:

- **Bootstrap (HTTP):** reads Locations, Things, Sensors, ObservedProperties and Datastreams
  from the source in dependency order and creates them on the target. Source IDs are
  rewritten to target IDs, and the mapping is kept in a state store.
- **Scoped sync:** limit the copy to a subgraph via an STA navigation path
  (`ObservedProperties(5674)/Datastreams`) or an OData filter. Dependencies are resolved
  automatically.
- **Live data (MQTT):** new Observations, entity changes and newly created entities are
  streamed from the source's MQTT broker to the target.
- **Pluggable parts:** state store (in-memory or Redis), transformers, target (STA HTTP or
  Kafka), and Basic/OAuth authentication. Multiple instances can share the work through a
  Redis-based federation coordinator.
- **Observability:** optional OpenTelemetry spans and metrics (API only, no SDK dependency).

Historical Observations are not copied during bootstrap; only Observations published after
the MQTT subscription starts are forwarded.

The code follows a hexagonal layout (`models`, `ports`, `adapters`, `pipeline`, `config`,
`transforms`) under `src/aio_sensorthings/`.

## Installation

Requires Python 3.13 or 3.14.

```bash
pip install "aio-sensorthings[all]"   # or pick extras: http, mqtt, redis, kafka, otel
```

## Usage

Scoped copy of all Datastreams of one ObservedProperty from a remote server to a local
FROST-Server, followed by live MQTT forwarding (state kept in Redis):

```python
import asyncio

import redis.asyncio as aioredis

from aio_sensorthings import (
    EntitySyncer,
    IdentityTransformer,
    ProcessRunner,
    RedisStateStore,
    STAClient,
    STAEntityType,
    STAHttpSource,
    STAHttpTarget,
    STAMqttSubscriber,
    StandaloneCoordinator,
)
from aio_sensorthings.config import MqttSettings


async def main() -> None:
    redis_client = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
    mqtt = MqttSettings(broker_host="iot.hamburg.de", broker_port=1883, qos=1, clean_session=False)

    async with (
        STAClient("https://iot.hamburg.de") as source_client,
        STAClient("http://localhost:8080/FROST-Server") as target_client,
    ):
        syncer = EntitySyncer(
            source=STAHttpSource(source_client),
            target=STAHttpTarget(target_client),
            state=RedisStateStore(redis_client, key_prefix="demo"),
            transformer=IdentityTransformer(),
            entity_types=[
                STAEntityType.LOCATION,
                STAEntityType.THING,
                STAEntityType.SENSOR,
                STAEntityType.OBSERVED_PROPERTY,
                STAEntityType.DATASTREAM,
            ],
            sync_by="name",
            source_mqtt=STAMqttSubscriber(mqtt),
            coordinator=StandaloneCoordinator(),
        )

        runner = ProcessRunner(
            syncer,
            bootstrap_kwargs={
                "entry_point": STAEntityType.DATASTREAM,
                "nav_path": "ObservedProperties(5674)/Datastreams",
            },
        )
        await runner.run()  # bootstrap, then subscribe until SIGINT/SIGTERM


asyncio.run(main())
```

More examples (federation, OpenTelemetry) are in [examples/](examples/).

---

## Entity mapping DSL

`MappingTransformer` maps fully expanded STA payloads to typed Pydantic models. The
mapping is a YAML or JSON document: one section per entity type, one named block per
output, one rule per target field.

```yaml
version: 1
DATASTREAM:
  CountingField:
    when: {from: properties.art, equals: "Zählfeld"}   # output condition
    name: Thing.properties.assetID                      # shorthand for {from: ...}
    unit: properties.aggregationMethod
    method: {const: infrared}
    traffic_node_arm.direction:
      from: Thing.properties.richtung
      regex: '^Richtung (?P<num>\d+)$'
      group: num
    id: {is_id: true, id_source: [Thing.properties.assetID, properties.knotenName]}
    vehicle_class:
      from: properties.typ
      when: {from: properties.typ, regex: "^(Kfz|Rad)$"}  # field condition
```

**Field rules**

- A scalar is a source path (`Thing.Locations[0].location`) and is shorthand for `from`.
- `const` sets a literal value.
- `regex` post-processes the source value; `group` picks a named or numbered group;
  `match_all` returns all matches as a list.
- `is_id` with `id_source` (list of paths) and `id_separator` (default `-`) joins several
  sources into one ID.
- Nested blocks and dotted keys are equivalent (`a: {b: x}` is `a.b: x`). A target field
  named like a rule key (`from`, `const`, `when`, ...) must use the dotted form.

**Conditions (`when`)**

`when` is optional on an output and on a single field. Without it, the output or field is
always mapped.

| Key | Meaning |
|---|---|
| `from` | Source path to test (required, never inherited from the field). |
| `equals` | Exact match against a scalar (str, int, float, bool). |
| `regex` | Regex searched in the value. |
| `negate` | Optional, inverts the result. |

Exactly one of `equals` or `regex` is required. Source values are converted with `str()`
before matching.

- **Output condition:** evaluated first. If it is false, the output is skipped and none of
  its field rules are evaluated.
- **Field condition:** if false, the field is left out, so model defaults apply. A required
  field that is left out fails validation, and the output is skipped with a warning.
- **Missing source:** if the source path resolves to nothing, the condition is false, even
  with `negate`. A warning is logged once per payload and path.
- Source paths in conditions are included in the derived `$expand` (`required_expands`).

## Prerequisites

For development, install one of the following package managers **once** on your machine
before proceeding:

| Package manager | Install |
|---|---|
| **pixi** *(default)* | `curl -fsSL https://pixi.sh/install.sh \| bash` — [pixi.sh](https://pixi.sh) |
| **poetry** | `pipx install poetry` — [python-poetry.org](https://python-poetry.org) |

For Docker workflows you also need [Docker](https://docs.docker.com/get-docker/).

---

## Development setup

### pixi

Pixi manages Python, all conda-forge and PyPI dependencies, and the virtual environment in a
single tool. The environment lives at `.pixi/envs/` and is **not** committed to git.

```bash
# Install all dependencies (creates the environment from pixi.lock)
pixi install

# Open a shell with dev tools available
pixi shell -e dev

# Or run a single task without activating the shell
pixi run -e dev test
pixi run -e dev lint
pixi run -e dev format
pixi run -e dev typecheck
```

Add a new dependency:

```bash
pixi add numpy                     # conda-forge, runtime
pixi add --pypi requests           # PyPI, runtime
pixi add --feature dev pytest-mock # dev-only
```

Update all dependencies and regenerate the lockfile:

```bash
pixi update
git add pixi.lock && git commit -m 'chore: update pixi lockfile'
```

## Common development tasks

All tasks are available via `make`. Run `make` with no arguments to execute
`lock → build-image → run-local`.

| Command | What it does |
|---|---|
| `make test` | Run the test suite |
| `make lint` | Lint with ruff |
| `make format` | Auto-format with ruff |
| `make typecheck` | Type-check with mypy |
| `make lock` | Update the dependency lockfile |
| `make build-image` | Build the Docker image |
| `make push-registry` | Push the image to the container registry |
| `make run-local` | Start the app locally via docker-compose |
| `make build-docs` | Build the Jupyter Book documentation |
| `make clean-docs` | Remove the documentation build output |

---

## Pre-commit hooks

Code quality checks run automatically on every `git commit`. Install the hooks once after cloning:

```bash
pre-commit install
```

Run all hooks manually at any time:

```bash
pre-commit run --all-files
```

The hooks cover: **ruff** (lint + auto-fix), **ruff-format**, **mypy**, and standard
file hygiene (trailing whitespace, YAML validity, merge-conflict markers, large files).

---

## Docker

Copy `.env.example` to `.env` and fill in the values (copier does this automatically on
first scaffold):

```bash
cp .env.example .env
```

| Command | What it does |
|---|---|
| `make build-image` | Builds `${CONTAINER_REGISTRY}/${IMAGE_NAME}:${IMAGE_TAG}` |
| `make run-local` | Runs the image via `docker-compose-dev.yaml` |
| `make push-registry` | Pushes to the container registry |

Key Docker files:

| File | Purpose |
|---|---|
| `Dockerfile` | Multi-stage build (adapts to the chosen package manager) |
| `docker-compose-build.yaml` | CI/CD image build |
| `docker-compose-dev.yaml` | Local dev/test run with secrets and env vars |
| `docker-compose.yaml` | Production deployment spec |

---

## Project structure

```
├── src/aio_sensorthings/                    # importable package
├── tests/                                   # pytest test suite
├── docs/                                    # Jupyter Book documentation
├── data/                                    # local data
├── assets/                                  # images, figures
├── reports/                                 # generated reports and outputs
├── examples/                                # usage examples
├── scratch/                                 # throwaway experiments (not tested)
└── secrets/                                 # local credentials — never committed
```

---

## VS Code setup

Recommended extensions (install via the Extensions panel or `.vscode/extensions.json`):

| Extension | Purpose |
|---|---|
| `ms-python.python` + `ms-python.pylance` | Python language support |
| `charliermarsh.ruff` | Linting and formatting (replaces black + isort) |
| `ms-python.mypy-type-checker` | Type checking |

Formatting on save and import sorting are pre-configured in `.vscode/settings.json`.

---

## Updating the template

To pull in improvements from the project template into this project:

```bash
copier update . --trust
```

Copier merges template changes with your local edits. Review the diff carefully before committing.