elwood-spatial 0.1.0

Information-theoretic outlier detection for spatial time series networks

elwood-spatial

Information-theoretic outlier detection for spatial networks.

Installation

pip install elwood-spatial

Quick Start

import elwood_spatial as es

# 1. Define discrete bins for your measurements (domain specific)
bins = es.BinSpec.from_tuples([(0, 50), (51, 100), (101, 150), (151, 200)])

# 2. Sensor readings at a single timestep
values = {"sensor_1": 45, "sensor_2": 120, "sensor_3": 48, "sensor_4": 52}
bin_indices = {k: bins.bin_index(v) for k, v in values.items()}
# => {'sensor_1': 0, 'sensor_2': 2, 'sensor_3': 0, 'sensor_4': 1}

Inspecting Network Entropy

Inspect the entropy of the network to see how much agreement exists among measurements:

all_indices = list(bin_indices.values())

# Shannon entropy — low values mean most sensors agree
entropy = es.shannon_entropy(all_indices)
print(f"Network entropy: {entropy:.3f} bits")

# Probability distribution across bins
probs = es.bin_probabilities(all_indices)
print(f"Bin probabilities: {probs}")
# => {0: 0.5, 2: 0.25, 1: 0.25}

Viewing Information Content per Sensor

Information content tells you how "surprising" each measurements is relative to its neighbours. A rare bin assignment produces high information:

for sensor_id, bi in bin_indices.items():
    ic = es.information_content(bi, all_indices)
    bd = es.bin_deviation(bi, [v for k, v in bin_indices.items() if k != sensor_id])
    print(f"{sensor_id}: value={values[sensor_id]}, bin={bi}, "
          f"information={ic:.3f} bits, bin_deviation={bd:.3f}")

Detecting Outliers

The rule-based detector flags a measurement when all three conditions hold:

1. Information content >= θ (default 1.75)
2. Network entropy < S (default 1.75)
3. Bin deviation >= n_bins / β (default 3.5)

from elwood_spatial.detect import detect_outliers, PARAMS_OPERATIONAL

network = {
    "sensor_1": {"neighbors": ["sensor_2", "sensor_3", "sensor_4"], "weights": [1.0, 0.9, 0.8]},
    "sensor_2": {"neighbors": ["sensor_1", "sensor_3", "sensor_4"], "weights": [1.0, 0.9, 0.7]},
    "sensor_3": {"neighbors": ["sensor_1", "sensor_2", "sensor_4"], "weights": [0.9, 0.9, 0.8]},
    "sensor_4": {"neighbors": ["sensor_1", "sensor_2", "sensor_3"], "weights": [0.8, 0.7, 0.8]},
}

results = detect_outliers(values, bins, network, PARAMS_OPERATIONAL)
for sid, flagged in results.items():
    print(f"{sid}: {'OUTLIER' if flagged else 'normal'}")

Full Metrics at a Glance

Use compute_network_metrics to get entropy, information, and bin deviation for every device in one call:

metrics = es.compute_network_metrics(bin_indices)
for device_id, m in metrics.items():
    print(f"{device_id}: H={m['entropy']:.3f}  I={m['information']:.3f}  D={m['bin_deviation']:.3f}")

Air Quality Module

For the originally intended air quality application, this package provides presets:

from elwood_spatial.air_quality import AQI_BINS, AQI_MODIFIED_BINS, pm25_to_aqi

# Convert PM2.5 to AQI
pm25_values = [5.0, 35.4, 55.5, 150.0]
for pm in pm25_values:
    aqi = pm25_to_aqi(pm)
    bin_idx = AQI_MODIFIED_BINS.bin_index(aqi)
    label = AQI_MODIFIED_BINS.labels[bin_idx] if bin_idx >= 0 else "?"
    print(f"PM2.5={pm:.1f} → AQI={aqi} → bin {bin_idx} ({label})")

Fault Simulation

The examples/synthetic_faults.py utility shows creating realistic sensor faults (SPIKE, FLATLINE, GAIN, DROPOUT, NOISE) into a time series DataFrame to test how the detection system responds:

from examples.synthetic_faults import inject_fault, FaultType

df = inject_fault(df, unit_id="sensor_3", fault_type=FaultType.SPIKE, seed=42)
# Adds a 'perturb' column marking affected rows

See the examples/ notebooks for some use cases.

Documentation

Docs, interactive testbed, and guides: https://elwood-spatial.com/

Development

pip install -e ".[dev]"
pytest tests/ -v
ruff check src/

License

BSD-3-Clause