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

Prerequisites

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

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.

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

Key Docker files:

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

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.