EnergyChain is the Python developer infrastructure through which applications can read, normalize, remember, analyze and explain energy system observations.

The problem it solves

Energy APIs are fragmented across devices and providers. An application may need to understand watts versus watt-hours, battery state of charge, observation timestamps, equipment IDs, data trustworthiness, and which actions a user is permitted to take. EnergyChain supplies common developer contracts instead of forcing every app to reconstruct them.

What you can build

• Battery assistants: ask how much backup time remains under specific loads.
• Solar intelligence: combine weather irradiance and explicit PV assumptions to estimate generation.
• Energy analytics: summarize consumption, costs, statistical outliers and baseline forecasts.
• Fleet applications: observe multiple authorized sites without silently discarding missing readings.
• Advisory automation: construct repeatable, permission-checked workflows and planning tools.

Choose your path

1. New to EnergyChain? Install and complete the first ten-minute project.

2. Integrating data? Map CSV records, Home Assistant entities, or a custom connector.

3. Building AI apps? Study the tools, structured memory, and bounded agent runtime.

4. Operating an application? Start with permissions, data freshness, secrets and failure modes.

Five design principles

• Measured is not estimated. A source timestamp and quality label accompany readings.
• Tools are contracts. Deterministic calculations should not be delegated to an LLM.
• Memory is structured. Time-series records belong in data stores, not prompt strings.
• Authorization precedes execution. LLM output never grants permissions.
• Simulator first. Read-only integration precedes physical action.

Scope boundaries

This SDK is not a national grid controller, SCADA replacement or a packaged end-user dashboard. It provides reusable developer infrastructure for applications built around energy data and intelligent decision support. Production services must supply their own authenticated entry points, tenant isolation, operational monitoring and deployment infrastructure.

Requirements

• Python 3.10 or newer (tested in CI on 3.10–3.13).
• A shell, Git and a Python virtual environment.
• No paid LLM account is necessary for the built-in simulator and deterministic tools.

Install from the development branch

The expanded v0.1 API described here currently lives on a feature branch. Until a public distribution containing those changes is confirmed, install from that branch after GitHub authentication:

git clone https://github.com/Atanseiye/energychain.git
cd energychain
git checkout feature/v0.1-deep-energy-platform-20261026
python -m venv .venv
# macOS / Linux
source .venv/bin/activate
python -m pip install -e '.[dev]'
python -m pytest -q

For Windows PowerShell, activate the environment with:

.\.venv\Scripts\Activate.ps1
python -m pip install -e ".[dev]"
python -m pytest -q

Confirm imports

from energychain import EnergyChain
from energychain import Fleet, TelemetryIngestor
chain = EnergyChain.demo()
print(len(chain.connectors.assets()))  # 4 in the built-in demo

Troubleshooting

If import fails, ensure you activated the correct virtual environment and are running from the selected Git branch. If your company GitHub account does not have access to the private repository, request repository access rather than publishing the URL or exposing credentials.

Next

Continue to First application. No hardware or third-party credentials are required.

Build an energy assistant without hardware

Start from the deterministic hybrid-energy simulator. It supplies four assets representing solar, storage, office loads and backup generation.

from energychain import EnergyChain

chain = EnergyChain.demo()
developer = chain.principal(
    "quickstart-dev",
    allowed_sites={"demo-site"},
)

assets = chain.tools.invoke(
    "discover_assets", developer, "demo-site"
)
print([(asset.id, asset.kind.value) for asset in assets])

battery = chain.tools.invoke(
    "get_battery_state",
    developer,
    "demo-site",
    arguments={"asset_id": "demo-battery"},
)
print("Battery SOC:", battery["state_of_charge"].value)

estimate = chain.tools.invoke(
    "estimate_battery_runtime",
    developer,
    "demo-site",
    arguments={
        "battery_id": "demo-battery",
        "load_w": 900,
        "reserve_pct": 20,
    },
)
print("Estimated hours:", estimate["hours"])
print("Data quality:", estimate["quality"])

Understand the result

The runtime tool uses battery capacity, charge, reserve, an assumed efficiency and a fixed load. It returns an estimate, not a promise that equipment can run for that duration. In the simulator, source readings are labelled simulated.

Try a second tool

cost = chain.tools.invoke(
    "estimate_energy_cost",
    developer,
    "demo-site",
    arguments={
        "energy_wh": 2000,
        "price_per_kwh": 100,
    },
)
print(cost["cost"])  # 200.0 in the chosen currency

What just happened?

1. The connector registry located the registered site and asset.

2. The tool registry enforced site and permission rules.

3. The energy calculator applied unit-correct arithmetic.

4. The memory and tracer could record observations and execution events.

Production warning

chain.principal(...) is provided for local demonstrations only. Do not let untrusted callers self-assign site access or privileges in a hosted application.

Foundations → Integration → Intelligence → Operations

Use the documentation as a layered curriculum rather than a random reference catalog.

Foundation

Complete Installation, First application, Energy data model, and Tools API. You should be able to explain which objects represent a site, asset, reading and tariff.

Build with data

Complete Connector framework, CSV, Home Assistant, and Weather and solar. Focus on provenance, validation and freshness rather than integration count.

Build intelligence

Study Energy Memory, Agents and models, Energy Intelligence, Fleets, Workflows, and Skills. Keep statistics and equipment constraints deterministic.

Deploy responsibly

Read Security and actions, Testing and evaluation, Architecture, Operations, and Troubleshooting. An application based on a tested SDK is not automatically a field-certified energy control system.

Documentation conventions

• Working denotes code implemented on the expanded v0.1 feature branch.
• Experimental denotes a narrow API without production guarantees.
• Planned denotes architecture or roadmap only.
• Never supported by v0.1 denotes capabilities intentionally absent, such as unrestricted physical dispatch.

Python examples show public SDK imports where possible and modules where necessary. Time-series examples use timezone-aware ISO 8601 timestamps.

Reference architecture

Your application (identity, routes, operator UX)
  └── EnergyChain SDK
      ├── Energy Core: sites, assets, topology, readings
      ├── ConnectorRegistry: CSV / simulator / HA adapter
      ├── ToolRegistry: site-aware permissions and callable tools
      ├── Memory: readings / events / notes / preferences / episodes
      ├── EnergyContextBuilder: bounded scoped retrieval
      ├── FleetManager: authorized multi-site read views
      ├── EnergyAgent + BoundedEnergyAgent
      ├── EnergySkillRegistry: reusable instructions
      ├── Intelligence: forecasts / economics / optimization
      ├── TelemetryIngestor + in-process event bus
      ├── EnergyWorkflow + SQLite checkpoints
      └── ActionService: prepare first, dispatch disabled

The 15-component inventory

| --- | --- |

Important separation

The SDK is embedded inside the developer's application. It is not an HTTP service, identity server or production worker fleet. The host application owns authentication, API boundaries, authorization issuance, secrets, uptime and tenancy.

Dependency direction

Connector observations flow through normalized energy objects. Tools enforce permission boundaries and invoke domain calculators. Agents consume tools through a restricted contract, never through arbitrary Python execution. Structured memory stores evidence; prompt context is a bounded derivative of that evidence.

Deployment profile

The recommended v0.1 deployment is read-only, single-trust-domain integration with a known data source and explicit failure handling. See Operations and Security and actions for a realistic production integration checklist.

The common language of energy

The public API includes EnergySite, EnergyAsset, EnergyReading, EnergyEvent, Tariff, EnergyPlan, Metric, AssetKind and Capability. The deep v0.1 branch adds EnergyRelationship, EnergyTopology, EnergyConstraint, EnergyActionSpec and require_fresh.

from datetime import datetime, timezone
from energychain import (
    EnergySite, EnergyAsset, EnergyReading, AssetKind, Metric
)

site = EnergySite("home-01", "Home energy")
solar = EnergyAsset("roof-pv", "home-01", "Solar PV", AssetKind.SOLAR)
reading = EnergyReading(
    asset_id="roof-pv",
    metric=Metric.POWER_W,
    value=1850.0,
    timestamp=datetime.now(timezone.utc),
    source="verified-export",
    quality="measured",
)
print(reading.unit)  # W

Physical quantity ≠ unit

| --- | --- | --- |

Do not add unrelated power observations across generators, batteries and loads without defining a consistent boundary and sign convention.

Provenance and freshness

Every reading has a timestamp, a provider source and quality. Measured comes from a sensor or data export; estimated comes from an algorithm; simulated comes from test fixtures. A provider's last_updated does not mean your application received the value in real time.

Use require_fresh when decisions need a bounded data age. Treat stale or missing values as explicit failures instead of silently carrying forward the last good reading.

Topology and constraints

EnergyTopology supports declared relationships across registered assets. EnergyConstraint checks bounded numeric values. Neither is a substitute for a certified electrical power-flow solver.

Recommended contract

Store stable, provider-independent identifiers. Always specify your site's timezone and tariff currency in the host domain model. Keep conversion logic at connector boundaries and retain original source details for troubleshooting.

Connector contract

All connectors derive from EnergyConnector and expose a consistent interface:

• sites() -> list[EnergySite]
• assets(site_id) -> list[EnergyAsset]
• read(asset_id, metric) -> EnergyReading
• history(asset_id, metric, start, end) -> list[EnergyReading]

ConnectorRegistry rejects conflicting site and asset identifiers across registered sources and routes observations to their owning connector.

Register a connector

from energychain import EnergyChain, HybridSiteSimulator

sdk = EnergyChain()
sdk.add_connector("north", HybridSiteSimulator(site_id="north-site"))
sdk.add_connector("south", HybridSiteSimulator(site_id="south-site"))
print([site.id for site in sdk.connectors.sites()])

Connector design checklist

1. Define stable asset identities, typed metrics and canonical units.

2. Validate the provider's timestamp, entity ID, numeric value and status.

3. Never report a cached value as live without a freshness check.

4. Map errors into explicit failures; avoid returning plausible substitute readings.

5. Declare unsupported methods with UnsupportedCapability.

6. Never infer authorization from the fact that a device is reachable.

7. Require TLS, authenticated providers and secrets supplied by your host app.

8. Make credential rotation, rate limits and provider outages visible to operators.

Included adapters

| --- | --- | --- |

Extending the SDK

Subclass the connector, implement the four methods and add it through sdk.add_connector(). Do not import proprietary API credentials into the package or publish customer data as tests.

Integration maturity

The existence of the adapter class proves only software contract implementation. Vendor certification, manufacturer-specific accuracy and actual pilot validation require testing on real approved installations.

Prepare your data

The CSV adapter accepts required fields asset_id,metric,value,timestamp,source with an optional quality field.

asset_id,metric,value,timestamp,source,quality
office-load,power_w,1200,2026-10-26T10:00:00Z,meter-export,measured
office-load,power_w,900,2026-10-26T11:00:00Z,meter-export,measured

Load the export

from pathlib import Path
from energychain import (
    EnergyChain, CSVConnector, EnergySite,
    EnergyAsset, AssetKind
)

site = EnergySite("office", "Office")
load = EnergyAsset("office-load", "office", "Main load", AssetKind.LOAD)
connector = CSVConnector.from_csv(
    Path("readings.csv"),
    site=site,
    assets=[load],
)
sdk = EnergyChain()
sdk.add_connector("office-csv", connector)
user = sdk.principal("local-user", allowed_sites={"office"})

latest = sdk.tools.invoke(
    "get_asset_state", user, "office",
    arguments={"asset_id": "office-load", "metric": "power_w"},
)
print(latest.value, latest.timestamp, latest.quality)

Semantics

The latest row is the most recently recorded measurement. It is not proof of a live connection. CSV readings require offset-aware timestamps; invalid values or unknown metric names are rejected.

Production considerations

For untrusted data feeds, add upload validation and file size limits outside the SDK, identify the author of every export, reconcile duplicate records and establish file retention policies.

Next

Move to Tools API to retrieve history and derive consumption summaries.

Intended use

Use a Home Assistant installation to expose battery, solar or load sensor readings to a developer application through EnergyChain's read-only connector.

Example: map a load sensor

import os
from energychain import (
    EnergyChain, EnergySite, EnergyAsset,
    AssetKind, Metric, HAEntityMapping,
    HomeAssistantConnector
)

site = EnergySite("home", "My home")
load = EnergyAsset("load", "home", "House load", AssetKind.LOAD)

adapter = HomeAssistantConnector(
    site=site,
    mappings=[
        HAEntityMapping(
            asset=load,
            entity_id="sensor.house_power",
            metric=Metric.POWER_W,
            source_unit="W",
        )
    ],
    base_url="https://ha.example.org",
    token_provider=lambda: os.environ["HA_TOKEN"],
)
sdk = EnergyChain()
sdk.add_connector("home-assistant", adapter)

Read a mapped state

user = sdk.principal("local-user", allowed_sites={"home"})
reading = sdk.tools.invoke(
    "get_asset_state", user, "home",
    arguments={"asset_id": "load", "metric": "power_w"},
)
print(reading.value, reading.unit, reading.timestamp)

The underlying read-only endpoint is Home Assistant GET /api/states/{entity_id}. The adapter checks that the returned entity and source unit match the declared mapping, rejects nonnumeric values and converts kW to W where configured.

Important limitations

• Historical recorder queries are not implemented by this adapter.
• The adapter does not dispatch smart-device commands.
• HA_TOKEN must be managed by your application or secret manager.
• The SDK does not implement credential rotation, provider reconnection or tenant-aware secret storage.
• A changed sensor entity, unavailable state or stale timestamp must cause an explicit application-level response.

Operational recommendation

Begin with one site and one read-only metric. Compare results with the Home Assistant dashboard and, where relevant, equipment measurements before declaring accuracy.

Weather is an input to energy intelligence

The optional OpenMeteoIrradiance source returns hourly forecast points. Weather irradiance is measured in W/m²; PV generation is an estimated electrical output conditioned on installed capacity and a performance ratio.

from energychain import OpenMeteoIrradiance, estimate_pv_power

source = OpenMeteoIrradiance()
points = source.hourly(7.25, 5.19, days=2)
for point in points[:3]:
    print(point.timestamp, point.irradiance_w_m2)

estimate_w = estimate_pv_power(
    irradiance_w_m2=780,
    rated_capacity_w=5000,
    performance_ratio=0.78,
)
print(estimate_w)

Modeling assumptions

The helper is an idealized conversion, not a validated rooftop PV digital twin. Roof tilt, shading, ambient and module temperature, inverter efficiency, clipping and degradation are not comprehensively modeled.

External data governance

Verify the provider's current commercial use rights, API terms, allowed polling frequency, attribution and rate limits before running a monetized service. A forecast can be late, unavailable or revised; treat each case explicitly.

Advanced route

Compare the estimated generation to actual inverter measurements on a historical holdout. Save model inputs, parameters and outcomes separately so operators can understand forecast drift.

All intelligent behaviour starts with a tool contract

Call registered tools using:

result = sdk.tools.invoke(
    "tool_name",
    principal,
    "site-id",
    arguments={"parameter": "value"},
)

Authorization is checked before the tool handler runs. A caller-supplied site_id must match the principal's authorization and the asset's registered site. Tool arguments do not confer new rights.

Built-in tools in expanded v0.1

| --- | --- |

Discover available tools

from energychain import EnergyChain
sdk = EnergyChain.demo()
for tool in sdk.tools.describe():
    print(tool)

Calculate a power balance

user = sdk.principal("dev", allowed_sites={"demo-site"})
balance = sdk.tools.invoke(
    "calculate_energy_balance", user, "demo-site",
    arguments={
        "solar_w": 1800,
        "load_w": 2200,
        "generator_w": 0,
    },
)
print(balance["deficit_w"])  # 400

Error handling

Expect ValidationError, AssetNotFound, PermissionDenied, ApprovalRequired and UnsupportedCapability rather than swallowing provider failures. Distinguish missing data from legitimate numeric zero.

Extensibility

ToolRegistry.register(ToolSpec(...)) supports application-specific tools. Treat every custom tool as a security boundary; no text-model response should be able to register a privileged tool.

Why energy memory is different from a chatbot history

A production energy app cannot rely on a conversation transcript to understand state of charge, recurring outages or energy cost. EnergyChain represents typed readings and episodes separately from text prompts.

The memory architecture

Observed readings ────> InMemoryEnergyMemory / SQLiteEnergyMemory
Energy events ────────> Event history
User preferences ─────> Keyed structured preferences
Operational episodes ─> InMemoryEpisodeStore / SQLiteEpisodeStore
                      └── EnergyContextBuilder
                           └── Bounded context for agents and tools

Persist preferences and readings

from energychain import SQLiteEnergyMemory

with SQLiteEnergyMemory("energy.db") as memory:
    memory.set_preference("demo-site", "reserve_pct", 25)
    print(memory.preferences("demo-site"))

The SQLite backend also supports add_reading, history, add_event, events, remember and recall. Its semantics are intended for locally trusted environments, not arbitrary tenants sharing one database file.

Episodic memory

EnergyEpisode records an event and evidence-backed outcome under a specific site. SQLiteEpisodeStore is a separate durable store with site-scoped retrieval and deletion support.

Use episodes to ask what happened previously and what followed, not to fabricate missing sensor data.

Scoped context retrieval

EnergyContextBuilder composes readings, notes, episodes and preferences into bounded results. Its output should never replace the source-of-truth history table.

Seven memory concepts

1. Working context: current bounded task state.

2. Asset memory: persisted equipment descriptions and metadata.

3. Time-series memory: historical measured/estimated readings.

4. Episodic memory: events, explanations and outcomes.

5. Preference memory: site-specific user rules.

6. Semantic knowledge: manuals, documents and retrieval — not yet a production vector-RAG implementation.

7. Procedural memory: reusable tool workflows and policies.

Advanced cautions

• Do not put every numeric reading in a vector database.
• Enforce ownership at the storage layer, not just in prompts.
• Establish retention, deletion, encryption, backups and audit rules in the host service.
• Handle chronological ordering and offset-aware timestamps.
• Distinguish prior observations from proposed future actions.
• SQLite is not a distributed, tenant-isolated managed time-series service.

Two runtime styles

EnergyAgent accepts a tool explicitly chosen by the developer. BoundedEnergyAgent can accept a restricted, structured tool plan from an injected text model.

Both use the same permission-checking ToolRegistry. Neither grants hardware privileges to an LLM.

Model-agnostic explicit agent

from energychain import EnergyChain, CallableTextModel

sdk = EnergyChain.demo()
principal = sdk.principal("dev", allowed_sites={"demo-site"})

model = CallableTextModel(
    lambda system, user, context: "Energy reading: " + context[:100]
)
agent = sdk.agent(model)
response = agent.run(
    "How is the battery doing?",
    principal=principal,
    site_id="demo-site",
    tool_name="get_battery_state",
    tool_arguments={"asset_id": "demo-battery"},
)
print(response.answer)

The callback shown is a simple local function to demonstrate the interface—not an LLM integration.

Bounded planner

sdk.planning_agent(model, max_calls=4, allowed_tools={...}) creates a planner that accepts only a strict finite JSON plan of allowed read/analysis tools. The SDK rejects fabricated tool names, attempts to call privileged execution tools, malformed plans and disallowed scopes.

Why bounded planning matters

A free-form LLM response is not a safe executable instruction. The system requires schema checks, allowlists, call limits and per-tool authorization.

Prompt injection considerations

Treat source text, manual content, free-form notes, agent summaries and external tool responses as untrusted. The bounded planner restricts tool access, but does not fully solve prompt injection in a larger application.

Production responsibilities

Your host service must provide user identity, model credentials, budget/rate limits, prompt/result logging policy and escalation rules. Human-readable explanations should identify assumptions and timestamps instead of fabricating source evidence.

What is not shipped

Multi-agent swarms, autonomously scheduled jobs, arbitrary recursive tool execution and field-certified agent-directed equipment control remain out of v0.1 scope.

A skill is a reusable procedure, not an independent model

EnergySkill stores a name, an instruction, and the set of tools appropriate for an energy task. EnergySkillRegistry validates that referenced tool names exist.

from energychain import EnergyChain

sdk = EnergyChain.demo()
for skill in sdk.skills.list():
    print(skill.name, skill.tools)

Built-in concept areas

The standard registry contains reusable battery, cost and fleet skill instructions. These help developers express domain-specific reasoning consistently while continuing to enforce tool-level permissions.

Authoring principles

• Make the task objective explicit.
• Declare exact tools and expected types.
• Never imply missing data or successful hardware actuation.
• Require the agent to state uncertainty and provenance.
• Preserve a separation between instructions and verified calculations.
• Keep parameterized calculations in Python functions, not natural-language guesses.

Advanced patterns

Compose skills with an allowlisted agent and a context builder. A skill can suggest reading battery SOC, estimating runtime and explaining reserve assumptions, but should never implicitly authorize altering equipment settings.

Not yet included

A hosted skills marketplace, verified third-party skill signatures, complex prompt versioning and a distributed agent skills runtime are future work.

Deterministic first

EnergyChain includes intentionally understandable numerical methods, not an undisclosed “AI optimization” black box.

Battery runtime

The approximate model is:

usable_Wh = capacity_Wh × max(SOC% - reserve%, 0) / 100 × efficiency
runtime_h = usable_Wh / load_W

Assumptions include constant load, efficiency and no future charging. It is not a manufacturer-certified safety calculation.

Demand forecasting

simple_average_forecast computes an arithmetic mean of same-metric observations, repeating that value for the requested horizon. It is a baseline, not a trained forecasting model.

Statistical anomaly screening

detect_outliers flags z-score outliers in consistent metric readings. Rare anomalies, seasonal patterns and changing operating regimes require additional validated models.

Power balance

from energychain import energy_balance, estimate_pv_power

balance = energy_balance(
    solar_w=2300,
    load_w=3100,
    generator_w=0,
)
print(balance.deficit_w)  # 800

pv_w = estimate_pv_power(
    irradiance_w_m2=700,
    rated_capacity_w=4000,
    performance_ratio=0.78,
)
print(pv_w)

Advisory load scheduling

optimize_load_schedule chooses among bounded load-start options across declared planning slots to minimize a simplified grid-cost objective while respecting site power constraints.

The result is advisory, not a physical schedule dispatched to equipment. Check slot timestamps, tariffs, base loads, flexible load durations and unit interpretation.

Evaluation

evaluate_forecast(actual, forecast) exposes MAE, RMSE, bias and MAPE-style diagnostics. chronological_mean_backtest uses a chronological training/test split instead of random temporal leakage.

Model governance

For real deployments, test seasonal holdouts, sensor dropout, tariff shifts, negative values, battery operating envelopes, equipment failures and out-of-distribution conditions. Do not make savings claims until independently measured against a credible baseline.

Fleets are groups of sites, not a superuser privilege

A Fleet identifies a set of registered sites. Fleet operations require the requesting principal to be authorized for all included sites before retrieving data.

from energychain import EnergyChain, HybridSiteSimulator, Fleet

sdk = EnergyChain.demo()
sdk.add_connector(
    "second",
    HybridSiteSimulator(site_id="site-two"),
)
sdk.fleets.register(
    Fleet("pilot-fleet", frozenset({"demo-site", "site-two"}))
)
principal = sdk.principal(
    "pilot-operator",
    allowed_sites={"demo-site", "site-two"},
)
snapshot = sdk.tools.invoke(
    "get_fleet_snapshot",
    principal,
    "demo-site",
    arguments={
        "fleet_id": "pilot-fleet",
        "metric": "power_w",
        "max_age_seconds": 3600,
    },
)
print(snapshot["observations"])
print(snapshot["failures"])

Data-quality behaviour

A fleet snapshot has a timestamp, metric, observations and a list of explicit site/asset failures. It deliberately does not compute an aggregate “fleet net power” across heterogeneous solar, loads, generators and batteries.

Scale and boundaries

The built-in manager is an in-process structure. A production fleet service should add tenancy, durable enrollment, site ownership verification, pagination, live telemetry pipelines and independent operations audit.

Common use cases

• Compare site data freshness and coverage.
• Build dashboards across multiple approved sites.
• Identify missing observations for operator follow-up.
• Perform read-only fleet-wide performance summaries.

A fleet is a convenient developer model; it does not imply privileged access to physical devices.

Two workflow mechanisms

EnergyWorkflow runs an ordered list of tool steps in a single process. SQLiteWorkflowCheckpoints records and reuses read/analysis workflow results to reduce accidental duplicated work after restarts.

Sequential pipeline

from energychain import EnergyChain, EnergyWorkflow, WorkflowStep

sdk = EnergyChain.demo()
operator = sdk.principal("developer", allowed_sites={"demo-site"})

flow = EnergyWorkflow(
    sdk.tools,
    [
        WorkflowStep("discover_assets"),
        WorkflowStep(
            "get_battery_state",
            {"asset_id": "demo-battery"},
        ),
    ],
)
result = flow.run(operator, "demo-site")
print(result.succeeded, len(result.outputs))

Checkpoint semantics

SQLite checkpoints bind a workflow run ID to its principal, site and plan. Reusing a run ID with a different scope/plan is rejected rather than silently replaying an unrelated result.

Safety and reliability

• Read/analysis steps are appropriate for the current checkpoint mechanism.
• Persistent read checkpoints are not an exactly-once physical action guarantee.
• Handler failure stops the in-process workflow.
• A production task queue needs durability, retry policies, idempotency keys, poison-message handling and operations visibility.

Event-driven patterns

The local event bus can notify subscribed callbacks about an event. It is not a durable Kafka/MQTT substitute and should not be treated as a high-availability control plane.

A practical reference application

A scheduled host job can fetch observations from an approved connector, ingest validated readings, use tools to estimate reserve capacity, and create an operator-facing advisory notification. The host remains responsible for scheduling, network retries, delivery and authenticated recipients.

Ingestion principles

TelemetryIngestor writes normalized readings to the chosen memory backend after validating their site/asset association. Optional source event identifiers allow duplicate detection.

Event processing pipeline

External observation
  -> Provider adapter
  -> EnergyReading (source / metric / quality / timestamp)
  -> TelemetryIngestor
  -> Memory
  -> Optional EnergyEventBus listener
  -> Application-specific operator notification

Avoid subtle errors

• A duplicate delivery should not silently inflate integrated energy.
• A sensor returning 0 W is different from a sensor not reporting.
• Event timestamps from multiple timezones require normalization.
• A provider connection may remain available while its readings are stale.
• “Last written to memory” is not proof of “last physically observed”.

In-process bus

The EnergyEventBus supports subscribe(listener), unsubscribe(id), publish(event) and scoped recent event retrieval. A failing subscriber must not be mistaken for a successful external delivery.

Operational scaling limits

In v0.1 the event bus is in-memory and bounded. There is no durable broker, distributed deduplication table, cross-process fanout, regulated-retention audit service or automated incident queue.

Design a production ingestion layer

Use an authenticated ingestion API, schema validation, source signatures where available, backpressure, retries, quotas, durable event storage, observability and bounded retention. Export measurements to a dedicated time-series store for scale.

Foundational rule

An AI model can explain or recommend; it must never become the authority to control hardware.

EnergyChain separates permissions into READ, ANALYZE, PREPARE and EXECUTE. Tool scopes and asset-site ownership are checked independently of any model-produced text.

Principals and permissions

from energychain import (
    Principal, Permission, SafetyPolicy
)

operator = Principal(
    "operator-123",
    frozenset({"my-site"}),
    frozenset({Permission.READ, Permission.ANALYZE}),
)
policy = SafetyPolicy()
policy.authorize(operator, "my-site", Permission.READ)

Prepare is not dispatch

EnergyActionSpec declares an action's permitted parameters. ActionService supports bounded preparation, but physical dispatch is disabled by default and does not imply that a device accepted or completed a command.

Signed approvals

SignedApprovalAuthority can issue a short-lived signature bound to the principal, site, action name and exact arguments, with a one-time nonce.

Critical deployment warning: Its single-use memory is local to an authority process. A real distributed approval service needs shared atomic replay protection, independent issuer identity, robust authorization controls and key management.

Four outcomes to distinguish

1. Requested: a caller proposed an action.

2. Approved: a trusted authority authorized that exact action.

3. Accepted: a device/API acknowledged the command.

4. Verified: independent telemetry confirms the expected physical state.

The current SDK should not claim stages 3–4 for live equipment.

Security checklist

• Never expose sdk.principal() as a public identity endpoint.
• Keep approving and acting roles separate.
• Bind approval to every action argument, not merely the action name.
• Treat tool outputs and source documents as untrusted.
• Require electrical safety constraints and manufacturer-specific interlocks.
• Store audit evidence in an append-only or independently managed system.
• Prevent SSRF and credential leakage in custom connectors.
• Enforce service quotas and request deadlines.
• Threat-model time-of-check/time-of-use races and concurrent approvals.

Not in v0.1

Field-certified physical dispatch, distributed execution authorization, hardware interlocks, fail-safe device feedback loops and guaranteed command rollback.

Test pyramid

1. Unit tests: object validation, arithmetic, safety and memory invariants.

2. Adapter tests: mock provider payloads, faults, units and timestamps.

3. Integration tests: real approved data sources under read-only access.

4. Operations tests: reconnects, overload, retention, credential rotation and outages.

5. Field tests: measurement accuracy, hardware safety and independent action verification.

Run locally

python -m pip install -e '.[dev]'
python -m ruff check --isolated --select E4,E7,E9,F src tests scripts examples
python -m pytest -q
python -m build
python -m twine check --strict dist/*

Current engineering evidence

The expanded branch passed 58 GitHub CI tests across Python 3.10–3.13 on Linux and Python 3.12 on Windows at the verified revision. CI also exercises package validation and a clean-wheel smoke test.

Forecast evaluation

Use evaluate_forecast to compare paired actual and predicted numbers. For chronological data, use temporal train/test separation to avoid look-ahead leakage.

Acceptance gates for a pilot

• Operator-authorized connection and repeatable credentials.
• Correct asset IDs, units and measured time stamps.
• Data provenance and missing/stale telemetry handling.
• Isolation between independent operators or sites.
• Real-world accuracy measurements and baseline comparisons.
• Monitoring of failed provider calls, request latency and quota usage.
• Verified data rights, retention and privacy procedures.
• Clear escalation for any suggested changes to physical systems.

Understand what is deployable

An EnergyChain SDK is not itself a complete production service. The host app still needs HTTP APIs, authentication, organizational tenancy, observability infrastructure, secrets, scheduled workers and operational support.

A recommended read-only application topology

Web / mobile app
     |
API gateway + real authentication + rate limiting
     |
Application backend (EnergyChain SDK)
     |-- Tenant / site authorization service
     |-- Connector registry -> approved provider APIs
     |-- Tool registry -> calculators / forecasts
     |-- Scoped context and memory
     |-- Fleet data and advisory workflow
     |
Durable database / time-series store
Metrics + traces + alerting service
Secret manager + rotation and audit

Host responsibilities

| --- | --- | --- |

Operational checklist

• Separate staging and production environments.
• Pin versions, validate supply-chain provenance and scan dependencies.
• Keep API tokens in a secret manager and rotate them.
• Enforce TLS for all external endpoints; limit egress.
• Implement health checks, structured logs, tracing and SLIs/SLOs.
• Plan backups, tested restoration and data retention.
• Isolate tenants in databases and authorization code.
• Add a change-control process before enabling anything capable of physical impact.
• Maintain equipment/vendor support plans and escalation procedures.

Reference integration goal

Start with a read-only deployment connected to one authorized real energy data source. Measure freshness, ingestion reliability, unit accuracy and operator usefulness. Expand to more sites only after those signals meet documented thresholds.

Not included

EnergyChain Cloud, hosted multi-tenant services, production device command runtime and certified dispatch controls are not currently shipped.

SDK entrypoints

| --- | --- |

Public types

from energychain import (
    EnergyChain, EnergySite, EnergyAsset, EnergyReading,
    EnergyEvent, EnergyPlan, Tariff, Metric, AssetKind,
    EnergyConnector, ConnectorRegistry, CSVConnector,
    HybridSiteSimulator, InMemoryEnergyMemory,
    SQLiteEnergyMemory, EnergyEpisode, SQLiteEpisodeStore,
    Fleet, FleetManager, SafetyPolicy, Principal, Permission,
    SignedApprovalAuthority, EnergyAgent, BoundedEnergyAgent,
    EnergyWorkflow, WorkflowStep, SQLiteWorkflowCheckpoints,
    TelemetryIngestor, EnergyEventBus,
    HomeAssistantConnector, HAEntityMapping,
    OpenMeteoIrradiance,
)

Internal package map

energychain/
  client.py
  models.py
  security.py / approval.py / actions.py
  intelligence.py / planning.py / evaluation.py
  fleets.py / telemetry.py / skills.py
  workflows.py / checkpoints.py
  tools/ base.py / builtin.py / extended.py
  connectors/ base.py / registry.py / simulator.py / csv.py
  integrations/ home_assistant.py / open_meteo.py
  memory/ base.py / sqlite.py / in_memory.py / episodes.py / context.py
  agents/ runtime.py / model.py / planner.py

Common errors

| --- | --- |

Source of truth

The implementation and tests in the feature branch govern precise signatures. These docs describe that branch; until the PR is merged and released, do not interpret every page as a promise about a public PyPI distribution.

Import errors

Symptom: ModuleNotFoundError: energychain

Activate the virtual environment used to install the SDK. Confirm Python and pip refer to the same interpreter:

python -m pip show energychain
python -c "import energychain; print(energychain.__version__)"

Missing deep-v0.1 APIs

If Fleet, HomeAssistantConnector or BoundedEnergyAgent cannot be imported, check that you installed the expanded development branch rather than the earlier alpha source or an unrelated PyPI package.

Forbidden site access

Symptom: PermissionDenied

The principal's allowed site list does not include the requested site, or the asset belongs to a different registered site. Do not fix this by assigning wildcard access to unknown callers; use the trusted identity provider and tenant authorization service.

Missing data or stale readings

An observation may be old even when successfully returned. Check reading.timestamp, reading.quality, reading.source and the configured freshness threshold. Treat missing data as failure, not zero.

Home Assistant failure

Verify the HTTPS origin, valid token, exact entity ID, current unit attribute and whether the provider returns a numeric state. This connector intentionally does not expose historical recorder data.

Forecast looks constant

The included demand forecast is an arithmetic mean baseline. A flat horizon is expected. Train and validate an external forecasting model for stronger predictive accuracy.

Unapproved action

Symptom: ApprovalRequired

Execution requires exact trusted authorization. An old unsigned Approval object cannot substitute for a verified signed approval under the strengthened safety design. Do not disable execution approval checks to bypass this.

CI and release

Run local pytest, ruff, python -m build and twine check. Confirm correct distribution metadata and clean-wheel imports. An open draft pull request is not a published stable release.

Need help?

Start with the architecture, security model or Pull Request #2.

Current release boundaries

The expanded v0.1 feature branch has passed CI, including tests, wheel build and clean installation checks. That verifies packaging and local behaviour—not an external production deployment or real energy hardware safety.

Ready for developer exploration

• Typed energy data, source quality and units.
• Simulator, CSV, and integration adapters.
• A site-scoped tool registry with deterministic calculations.
• Structured local memory, episodes and bounded context.
• Fleet snapshots with explicit partial failures.
• Read-only model-directed tool planning.
• Load-scheduling advice and numerical constraints.
• Local workflows, telemetry, traces and evaluation tools.

Not yet validated for live operators

• Credentialed real-site data sources with independent measurements.
• Manufacturer-specific power/equipment semantics.
• Multi-tenant authorization and managed secrets.
• Durable distributed telemetry and worker orchestration.
• Production service SLAs, fault injection and disaster recovery.
• Certified physical-device actuation and independent outcome verification.

Deployment milestones

Milestone A — Pilot read-only integration: connect one authorized site; validate freshness, units and operator utility.

Milestone B — Operational data plane: authenticated APIs, tenant isolation, durable storage, broker, audit, retries and monitoring.

Milestone C — Intelligence evaluation: stronger models with time-based backtests, drift monitoring and measured performance.

Milestone D — Controlled actions: independent approval issuer, equipment interlocks, device acknowledgment, verified outcomes and hazard analysis.

Package publishing

GitHub and PyPI are independent. A private GitHub project can publish a publicly downloadable wheel; release licensing and PyPI Trusted Publishing must be explicitly approved and configured. Use the official repository release checks before claiming pip install energychain delivers the expanded implementation.