faketelemetry 0.4.1

Realistic fake telemetry streams for testing, simulation & IoT development.
Zero dependencies. Pure Python 3.7+.

PyPI • GitHub • Quickstart • CLI • Features • API Reference

Quickstart

pip install faketelemetry

from faketelemetry import TelemetryGenerator, WaveformType

gen = TelemetryGenerator(WaveformType.SINE, frequency=1.0, amplitude=5.0)

# Real-time stream (blocking, with sleep)
for ts, val in gen.stream(sampling_rate=10, duration=2):
    print(ts, round(val, 2))

# Instant batch (no sleeping -- great for tests & pipelines)
points = gen.batch(num_samples=100, sampling_rate=50)

Features

Installation

pip install faketelemetry

Or from source:

git clone https://github.com/adkvi/faketelemetry.git
cd faketelemetry
pip install -e .

Requirements: Python 3.7+ — no external dependencies.

CLI

After installing, the faketelemetry command is available in your terminal (or use python -m faketelemetry).

Generate from a sensor preset

faketelemetry preset temperature -n 10 --rate 1
faketelemetry preset cpu_usage -n 100 --rate 10 -f csv -o cpu.csv

Generate from a waveform

faketelemetry generate -w sine --freq 2 --amp 5 --offset 10 -n 50 --rate 10
faketelemetry generate -w square --duty-cycle 0.3 -n 20 -f json

Multi-channel output

faketelemetry multi temperature,pressure,cpu_usage -n 100 --rate 10 -f csv

Real-time streaming to stdout

faketelemetry stream --preset heart_rate --rate 2 --duration 30
faketelemetry stream -w sine --rate 10       # Ctrl+C to stop

Generate from a saved config

faketelemetry from-config config.json -n 500 -f ndjson -o data.ndjson

List available presets, waveforms, noise types

faketelemetry list presets
faketelemetry list waveforms
faketelemetry list noise
faketelemetry list anomalies

Summary statistics

faketelemetry describe --preset temperature -n 1000 --rate 10
faketelemetry describe -w sine -n 5000 --rate 100

Output formats

All batch commands support --format / -f with: table (default), csv, json, ndjson, influxdb.

Other flags

Python API Examples

Noise Injection

Six noise distributions, all with optional seed for reproducibility.

from faketelemetry import TelemetryGenerator, WaveformType, NoiseInjector, NoiseType

# Gaussian noise
gen = TelemetryGenerator(
    WaveformType.SINE,
    noise_injector=NoiseInjector(noise_level=0.5, noise_type=NoiseType.GAUSSIAN, seed=42),
)

# Brownian (random-walk) noise
gen = TelemetryGenerator(
    WaveformType.SINE,
    noise_injector=NoiseInjector(noise_level=0.1, noise_type=NoiseType.BROWNIAN),
)

# Quantization noise (simulates low-resolution ADC)
gen = TelemetryGenerator(
    WaveformType.SINE,
    amplitude=5.0,
    noise_injector=NoiseInjector(noise_type=NoiseType.QUANTIZATION, quantization_step=0.25),
)

Anomaly / Fault Injection

Stack multiple anomaly behaviours -- each fires independently per sample.

from faketelemetry import (
    TelemetryGenerator, WaveformType,
    AnomalyInjector, AnomalyConfig, AnomalyType,
)

gen = TelemetryGenerator(
    WaveformType.SINE,
    frequency=1.0,
    amplitude=5.0,
    anomaly_injector=AnomalyInjector(
        configs=[
            AnomalyConfig(AnomalyType.SPIKE, probability=0.05, magnitude=20.0),
            AnomalyConfig(AnomalyType.DRIFT, probability=1.0, drift_rate=0.01),
            AnomalyConfig(AnomalyType.DROPOUT, probability=0.02, duration_samples=5),
        ],
        seed=42,
    ),
)

values = gen.batch_values(num_samples=500, sampling_rate=100)

Sensor Presets

19 realistic sensor simulators, each returns a ready-to-use TelemetryGenerator.

from faketelemetry.presets import SensorPreset

temp    = SensorPreset.temperature(seed=1)        # ~22 C +/- 3 C
press   = SensorPreset.pressure(seed=1)            # ~1013 hPa
humid   = SensorPreset.humidity(seed=1)            # 0-100 %
battery = SensorPreset.battery_voltage(seed=1)     # 4.2 V decaying
rpm     = SensorPreset.rpm(seed=1)                 # ~800 RPM idle
cpu     = SensorPreset.cpu_usage(seed=1)           # 0-100 %
latency = SensorPreset.network_latency(seed=1)     # ~25 ms
hr      = SensorPreset.heart_rate(seed=1)          # ~72 BPM
accel   = SensorPreset.accelerometer(seed=1)       # +/- 1 g
gps     = SensorPreset.gps_coordinate(seed=1)      # wandering lat/lon
light   = SensorPreset.light_lux(seed=1)           # ~400 lux
co2     = SensorPreset.co2_ppm(seed=1)             # ~420 ppm
wind    = SensorPreset.wind_speed(seed=1)           # ~12 m/s
mem     = SensorPreset.memory_usage(seed=1)         # 0-100 %
disk    = SensorPreset.disk_io(seed=1)              # MB/s bursts
flow    = SensorPreset.flow_rate(seed=1)            # ~50 l/min
vib     = SensorPreset.vibration(seed=1)            # mm/s
ph      = SensorPreset.ph_level(seed=1)             # 0-14
rssi    = SensorPreset.signal_strength(seed=1)      # -120..0 dBm

data = temp.batch(num_samples=1000, sampling_rate=10)

Phase, Duty Cycle, Damping & Clamping

Fine-tune signal shape and bounds.

import math
from faketelemetry import TelemetryGenerator, WaveformType

gen = TelemetryGenerator(
    WaveformType.SINE,
    frequency=2.0,
    amplitude=10.0,
    phase=math.pi / 4,   # 45-degree phase shift
    damping=0.3,          # exponential decay envelope
    clamp_min=-5.0,       # hard floor
    clamp_max=5.0,        # hard ceiling
)

values = gen.batch_values(200, sampling_rate=100)

Multi-Channel with Named Channels

Synchronise multiple signals on the same time base.

from faketelemetry import TelemetryGenerator, WaveformType, MultiChannelTelemetryGenerator

multi = MultiChannelTelemetryGenerator({
    "temperature": TelemetryGenerator(WaveformType.SINE, frequency=0.1, offset=22.0, name="temp"),
    "pressure":    TelemetryGenerator(WaveformType.COSINE, frequency=0.05, offset=1013.0, name="press"),
})

# Batch -> list of dicts (pandas-ready)
import pandas as pd
df = pd.DataFrame(multi.batch(num_samples=1000, sampling_rate=10))
print(df.head())

# Or raw arrays for plotting
arrays = multi.batch_arrays(num_samples=500, sampling_rate=50)

Data Export

Write to file or get back a string -- CSV, JSON, or NDJSON.

from faketelemetry import TelemetryGenerator, WaveformType, to_csv, to_json, to_ndjson

gen = TelemetryGenerator(WaveformType.SINE, name="sensor_a")
data = gen.batch(num_samples=500, sampling_rate=100)

# To file
to_csv(data, "output.csv")
to_json(data, "output.json")
to_ndjson(data, "output.ndjson")

# To string
csv_string  = to_csv(data)
json_string = to_json(data)

Signal Composition

Combine generators with +, -, * to build complex signals.

from faketelemetry import TelemetryGenerator, WaveformType, compose

# Sum of harmonics (approximate square wave via Fourier series)
fundamental = TelemetryGenerator(WaveformType.SINE, frequency=1.0, amplitude=1.0)
harmonic3   = TelemetryGenerator(WaveformType.SINE, frequency=3.0, amplitude=1/3)
harmonic5   = TelemetryGenerator(WaveformType.SINE, frequency=5.0, amplitude=1/5)
approx_square = fundamental + harmonic3 + harmonic5

# Amplitude modulation
carrier  = TelemetryGenerator(WaveformType.SINE, frequency=100.0)
envelope = TelemetryGenerator(WaveformType.SINE, frequency=1.0, amplitude=0.5, offset=0.5)
am_signal = carrier * envelope

# Scalar offset
biased = TelemetryGenerator(WaveformType.SINE) + 10.0

# compose() for many generators
signal = compose(fundamental, harmonic3, harmonic5)  # same as +
values = signal.batch_values(500, sampling_rate=200)

Async Streaming

Native async for support for asyncio applications.

import asyncio
from faketelemetry import TelemetryGenerator, WaveformType

async def main():
    gen = TelemetryGenerator(WaveformType.SINE, frequency=1.0, amplitude=5.0)
    async for ts, val in gen.astream(sampling_rate=10, duration=3):
        print(f"{ts}: {val:.2f}")

asyncio.run(main())

Protocol Formatters

Output data in InfluxDB, MQTT, or Prometheus format -- ready for ingestion.

from faketelemetry import TelemetryGenerator, WaveformType
from faketelemetry.formatters import to_influxdb, to_mqtt_json, to_prometheus

gen = TelemetryGenerator(WaveformType.SINE, name="temp")
data = gen.batch(10, sampling_rate=1)

# InfluxDB line protocol
for line in to_influxdb(data, measurement="sensors", tags={"room": "lab"}):
    print(line)
# sensors,room=lab value=0.123 1706745600000000000

# MQTT JSON payloads
for payload in to_mqtt_json(data, sensor_id="temp-01"):
    print(payload)
# {"timestamp": "...", "value": 0.123, "sensor_id": "temp-01"}

# Prometheus exposition format
print(to_prometheus(data, metric_name="room_temp", labels={"room": "lab"}))

Statistics

Built-in summary stats -- no pandas required.

from faketelemetry import TelemetryGenerator, WaveformType, describe, describe_values

gen = TelemetryGenerator(WaveformType.SINE, amplitude=5.0)
data = gen.batch(1000, sampling_rate=100)

stats = describe(data)
print(stats)
# {'count': 1000, 'mean': 0.01, 'std': 3.54, 'min': -5.0, 'max': 5.0,
#  'range': 10.0, 'p25': ..., 'p50': ..., 'p75': ..., ...}

# Or from raw values
stats = describe_values(gen.batch_values(1000, sampling_rate=100))

Config Serialization

Save and replay generator setups as JSON.

import json
from faketelemetry import TelemetryGenerator, WaveformType, NoiseInjector

gen = TelemetryGenerator(
    WaveformType.SINE, frequency=2.0, amplitude=5.0,
    noise_injector=NoiseInjector(noise_level=0.3), name="sensor_x",
)

# Save config
config = gen.to_config()
json.dump(config, open("config.json", "w"))

# Reload later
gen2 = TelemetryGenerator.from_config(json.load(open("config.json")))

Reproducible Output

Same seed = identical data every run.

from faketelemetry import TelemetryGenerator, WaveformType, NoiseInjector

gen = TelemetryGenerator(
    WaveformType.SINE,
    noise_injector=NoiseInjector(noise_level=0.5, seed=42),
    seed=42,
)

run1 = gen.batch_values(100, sampling_rate=50)
gen.reset()
run2 = gen.batch_values(100, sampling_rate=50)
# run1 == run2

DataFrame Integration

import pandas as pd
from faketelemetry import TelemetryGenerator, WaveformType

gen = TelemetryGenerator(WaveformType.SINE, frequency=1.0, name="vibration")

df = pd.DataFrame(gen.to_dict(num_samples=1000, sampling_rate=100))
print(df.describe())

API Reference

Core Classes

Enums

Export & Formatting

Statistics

Testing

python -m pytest tests/ -v

Contributing

See CONTRIBUTING.md for guidelines.

Support

For questions or issues, please open a GitHub Issue.

License

MIT -- see LICENSE for details.