aio-sensorthings

An asyncio Python library and SDK for OGC SensorThings API (STA) servers such as 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.

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):

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/.

---

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.

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
poetry	pipx install poetry — python-poetry.org

For Docker workflows you also need 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.

# 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:

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:

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:

pre-commit install

Run all hooks manually at any time:

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):

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:

copier update . --trust

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