meterdatalogic

meterdatalogic is a lightweight Python package that provides data transformation, validation, and analytics logic for customer interval meter data.

• Canonical Data Shape — normalise datasets to a consistent schema for reliable analytics.
• Small, Composable Modules — ingest, validate, transform, summary, pricing, scenario.
• Framework-Agnostic — works in Django, FastAPI, notebooks, or jobs.
• Plot-Ready Outputs — Polars DataFrames or JSON-ready dicts.
• Self-Validating — schema checks for tz-aware, sorted, duplicate-free data.
• Optimised for interval energy data — ToU, demand windows, tariff calculation.

Requirements

• Python: 3.12+
• polars: >=1.40.1
• nemreader: >=1.0.0

Timezone Handling

All timestamps in the canonical schema are tz-aware. The default timezone is Australia/Brisbane (no DST). You can specify any valid timezone during ingest:

df = ml.ingest.from_dataframe(raw_df, tz="Australia/Sydney")  # DST-aware

Key principles:

• Input data with naive timestamps is localized to the specified timezone
• DST transitions are handled correctly (gaps and overlaps)
• All operations preserve timezone information
• Output timestamps remain tz-aware

Documentation

Comprehensive documentation is available in the docs/ folder:

• Release Workflow - Release workflow guide
• Data Structure Format - Data structure format
• Examples & Use Cases - Practical recipes
• API Reference - Complete API documentation

Project Structure

meterdatalogic/
  __init__.py       # Public API exports
  types.py          # Shared type definitions (CanonFrame, Plan, etc.)
  py.typed          # Type checking marker
  core/             # Core data structures and transformations
    __init__.py
    canon.py        # Canonical schema definitions and constants
    exceptions.py   # CanonError exception
    transform.py    # Aggregation, ToU binning, profile generation
    types.py        # Core type definitions
    utils.py        # Helper functions
  io/               # Data input/output and validation
    __init__.py
    ingest.py       # Data loading (NEM12, CSV, DataFrame)
    validate.py     # Schema validation
    formats.py      # Format conversion (to/from JSON)
    types.py        # I/O type definitions
  analytics/        # Analysis and modeling
    __init__.py
    summary.py      # JSON-ready summaries
    pricing.py      # Tariff calculations
    scenario.py     # Solar/battery/EV modeling
    types.py        # Analytics type definitions
    insights/       # Pattern detection & recommendations
      __init__.py
      engine.py     # Insight generation orchestration
      config.py     # Configuration and thresholds
      types.py      # Insight type definitions
      evaluators_*.py # Evaluator functions
tests/              # Test suite
docs/               # Documentation

Module Overview

All modules are accessible via a clean, flat API:

import meterdatalogic as ml

# Data I/O
ml.ingest      # Load NEM12, CSV, or DataFrame to canonical format
ml.validate    # Enforce schema rules (tz-aware, sorted, unique timestamps)
ml.formats     # Convert between CanonFrame and JSON representations

# Core operations
ml.canon       # Canonical schema definitions and constants
ml.transform   # Aggregate by time/ToU, calculate profiles and peaks
ml.utils       # Helper functions for common operations

# Analytics
ml.summary     # Generate JSON-ready summaries for dashboards
ml.pricing     # Calculate billables and costs from tariff plans
ml.scenario    # Model solar PV, battery storage, and EV charging
ml.insights    # Automated pattern detection and recommendations

Internal organization: The codebase is organized into domain-based packages (core/, io/, analytics/) for maintainability, but users don't need to know about this internal structure.

Canonical Schema

Every dataset processed conforms to the canonical schema:

• Column t_start: tz-aware Datetime column, strictly increasing.
• Columns:
  • nmi: str (single site per frame).
  • channel: str (source register label, e.g., E1, B1).
  • flow: str (grid_import, controlled_load_import, grid_export_solar).
  • kwh: float (energy in the interval; import/export indicated by flow, not sign).
  • cadence_min: int (interval minutes, e.g., 30/15/5).

Conventions:

• Import (customer consumption) and export (PV feed-in) are separate flows.
• Default timezone is "Australia/Brisbane" unless specified.

Quick Start

1) Ingest

Normalise raw data to canonical form.

import meterdatalogic as ml

df = ml.ingest.from_dataframe(raw_df, tz="Australia/Brisbane")
ml.validate.assert_canon(df)  # raises CanonError on issues

2) Transform

Unified aggregation helpers.

# Daily energy by flow (wide columns)
daily = ml.transform.aggregate(df, freq="1D", groupby="flow", pivot=True)

# Monthly peak demand (MF 16:00–21:00) in kW
demand = ml.transform.aggregate(
  df,
  freq="1MS",
  flows=["grid_import"],
  metric="kW",          # derive kW from kWh using cadence
  stat="max",           # max within each monthly bucket
  out_col="demand_kw",
  window_start="16:00",
  window_end="21:00",
  window_days="MF",     # ALL | MF (Mon–Fri) | MS (Mon–Sun?)
)

# Time-of-Use bins (month + one column per band name)
bands = [
  ml.types.ToUBand("off","00:00","16:00",22.0),
  ml.types.ToUBand("peak","16:00","21:00",45.0),
  ml.types.ToUBand("shoulder","21:00","24:00",28.0),
]
tou = ml.transform.tou_bins(df, bands)

# Average-day profile and top hours
prof = ml.transform.profile(df)  # columns: slot, flows..., import_total
top = ml.transform.top_n_from_profile(prof, n=4)
print(top["hours"])  # e.g., ['18','19','20','21']

3) Summary

JSON-ready summary payloads for dashboards.

summary = ml.summary.summarise(df)
print(summary["meta"])     # start/end/cadence/days
print(summary["energy"])   # totals per flow

4) Pricing

Estimate monthly bills from interval data.

plan = ml.types.Plan(
    usage_bands=[
        ml.types.ToUBand("off","00:00","16:00",22.0),
        ml.types.ToUBand("peak","16:00","21:00",45.0),
        ml.types.ToUBand("shoulder","21:00","24:00",28.0),
    ],
    demand=ml.types.DemandCharge("16:00","21:00","MF",12.0),
    fixed_c_per_day=95.0,
    feed_in_c_per_kwh=6.0,
)

bill = ml.pricing.compute_billables(df, plan, mode="monthly")
cost = ml.pricing.estimate_costs(bill, plan)

cycles = [("2025-05-31", "2025-06-30"), ("2025-07-01", "2025-07-30")]
bill_cycles = ml.pricing.compute_billables(df, plan, mode="cycles", cycles=cycles)
bills = ml.pricing.estimate_costs(bill_cycles, plan, pay_on_time_discount=0.07, include_gst=True)

5) Scenarios (EV, PV, Battery)

Simulate EV charging, PV generation, and battery self-consumption, then optionally price the outcome.

ev = ml.types.EVConfig(daily_kwh=8.0, max_kw=7.0, window_start="18:00", window_end="22:00", days="ALL", strategy="immediate")
pv = ml.types.PVConfig(system_kwp=6.6, inverter_kw=5.0, loss_fraction=0.15, seasonal_scale={"01":1.05,"06":0.9})
bat = ml.types.BatteryConfig(capacity_kwh=10.0, max_kw=5.0, round_trip_eff=0.9, soc_min=0.1, soc_max=0.95)

result = ml.scenario.run(df, ev=ev, pv=pv, battery=bat, plan=plan)

Testing

# Run all tests
uv run pytest
# or
make test

# Run with coverage
uv run pytest --cov=meterdatalogic

# Run specific test file
uv run pytest tests/test_transform.py

Development

# Install dependencies and pre-commit hooks
make install

# Lint code
make lint
# or
uv run ruff check .

# Format code
uv run ruff format .

# Run pre-commit hooks manually
uv run pre-commit run --all-files