iot-telemetry

name

iot-telemetry

description

Python IoT telemetry simulation for hot air balloon fleet with Azure Event Hubs integration. USE FOR: python simulator, telemetry generator, balloon metrics, event hubs producer, pydantic telemetry schema, pytest fixtures, flight scenario, GPS trajectory, altitude model, JSON payload. DO NOT USE FOR: TypeScript MCP tools, KQL queries, React components, CopilotKit agent, Fabric IQ ontology, frontend styling.

Overview

The SkyNav IoT telemetry simulator generates realistic hot air balloon flight data in Python and streams it to Azure Event Hubs. It uses Pydantic v2 models for schema validation and pytest with hypothesis for property-based testing.

Core Concepts

Telemetry Data Model

Each telemetry record represents a point-in-time snapshot of a balloon's state:

Field	Type	Description
balloon_id	str	Unique fleet identifier (e.g., "B-001")
timestamp	datetime	UTC capture time
latitude	float	GPS latitude (-90 to 90)
longitude	float	GPS longitude (-180 to 180)
altitude_m	float	Meters above sea level
temperature_c	float	Envelope temperature in Celsius
fuel_pct	float	Remaining fuel (0-100%)
heading_deg	float	Compass heading (0-360)
vertical_speed_mps	float	Vertical speed in m/s
status	str	One of: preflight, airborne, descending, landed

Event Hubs Integration

• Producer pattern: Async EventHubProducerClient with batch sending
• Partition routing: Use balloon_id as partition key for ordered per-balloon processing
• Serialization: model.model_dump_json() → EventData wrapper
• Connection: pydantic-settings BaseSettings with SKYNAV_EVENTHUB_CONNECTION_STRING

Flight Scenario Generator

Scenarios define a complete flight profile as a sequence of phases:

1. Preflight — Ground-level warmup, fuel check
2. Ascent — Burner on, altitude increases, fuel burns faster
3. Cruise — Altitude maintenance with periodic burner pulses
4. Descent — Controlled cooling, altitude decreases
5. Landing — Final approach to target coordinates

Each phase has configurable duration, altitude targets, and fuel burn rates.

Common Patterns

Creating a Simulator Instance

from skynav.simulator import FleetSimulator
from skynav.config import SimulatorSettings

settings = SimulatorSettings()  # reads from SKYNAV_* env vars
simulator = FleetSimulator(
    fleet_size=5,
    emit_interval_s=settings.emit_interval,
    scenario="standard_flight",
)
async for batch in simulator.run():
    await send_telemetry(client, batch)

Testing Telemetry Schemas

@given(st.builds(BalloonTelemetry, balloon_id=st.just("B-TEST")))
def test_telemetry_is_serializable(record):
    json_str = record.model_dump_json()
    restored = BalloonTelemetry.model_validate_json(json_str)
    assert restored == record

When to Use

• Building or modifying the Python telemetry simulator
• Defining or updating Pydantic telemetry schemas
• Writing Event Hubs producer logic
• Creating flight scenario generators
• Writing pytest/hypothesis tests for telemetry

When NOT to Use

• Building TypeScript MCP server tools → use mcp skill
• Writing KQL queries for Eventhouse → use mcp skill
• Building React dashboard components → use reactjs skill
• Defining Fabric IQ ontology entities → use mcp skill