DEV Community: Oleksandr

How DER Twin Works — Architecture of an Open Source Energy Device Simulator

Oleksandr — Tue, 07 Apr 2026 12:34:37 +0000

This is Part 3 of my series on DER Twin. Part 1 covered why I built it. Part 2 showed how to use it for EMS testing. This one is about how it works inside.

If you want to contribute — add a new device type, extend the physics, or add a new protocol — this is the article to read first.

The Four Layers

DER Twin is built around a strict separation of concerns. Every component lives in exactly one layer and communicates only through defined interfaces.

Protocol layer — accepts Modbus connections and exposes register addresses. Never touches device state directly.

Controller layer — maps register reads and writes to device telemetry and commands. Owns the encoding/decoding logic.

Device layer — owns physics. Knows nothing about protocols. A BESS device doesn't know if it's being read via TCP or RTU.

Simulation engine — owns time. Calls step(dt) on every device on every tick.

External models — the environment. Irradiance, ambient temperature, grid frequency, grid voltage. Devices read from them, never write to them.

This separation is the most important architectural decision in the project. It's what makes the simulator extensible — you can add a new protocol without touching device code, and add a new device without touching protocol code.

The Simulation Engine

The engine is the heartbeat of the simulator. On each tick it advances every device by one time step:

async def step_once(self):
    dt = self.clock.step
    self.external_models.update(self.sim_time, dt)
    for controller in self.devices:
        controller.step(dt)
    self.clock.advance()

The clock has two modes:

real_time: true — the engine sleeps between ticks to match wall clock time. Use this for live operation where your EMS polls at a realistic rate.

real_time: false — no sleeping. The engine runs as fast as the CPU allows. This is what makes 20 minutes of simulation complete in 0.30 seconds.

The clock also has a start_time_h parameter — set it to 12.0 and the simulation starts at solar noon. All external models see the correct time from the first tick.

Device Models

Each device type follows the same interface:

class SimulatedDevice:
    def update(self, dt: float) -> None: ...
    def get_telemetry(self) -> ...: ...
    def apply_commands(self, commands: dict) -> dict: ...

update() advances the physics by dt seconds. get_telemetry() returns the current state snapshot. apply_commands() accepts a dict of register name → value and applies whatever the device understands.

BESS

The BESS model is built around energy integration. On each step it tracks SOC, enforces power limits, and applies ramp rate constraints:

Power is clamped to max_charge_kw and max_discharge_kw. A ramp rate limits how fast power changes when a new setpoint arrives — the BESS doesn't jump instantly to full power, as real hardware wouldn't. SOC is updated each step:

SOC += (power_kw / capacity_kwh) * (dt / 3600) * 100

Temperature from the ambient model affects available power. Grid voltage and frequency are reflected in telemetry and can trigger protection responses.

PV Inverter — DC/AC Separation

The PV model separates DC generation from AC conversion, mirroring the physical reality of a solar installation:

DC side (PVArrayModel) converts irradiance into available DC power using a NOCT-based thermal model. Cell temperature rises with irradiance — above 25°C the module loses roughly 0.4% per degree, which meaningfully affects output on hot days.

AC side (PVInverterModel) takes DC input and applies inverter efficiency, AC rating clamp, and curtailment. It also runs grid protection — if voltage or frequency moves outside safe limits the inverter trips and sets a fault code. The thermal model tracks inverter temperature as a function of conversion losses.

Both total_input_power (DC) and total_active_power (AC) are exposed in telemetry so your EMS can read both sides of the conversion. The panel doesn't know about the grid and the inverter doesn't know about irradiance — extending one doesn't require touching the other.

External Models

External models represent the environment that devices operate in. They're updated once per tick before devices are stepped, so every device sees a consistent environment snapshot within each step.

The irradiance model produces a sine curve between sunrise and sunset:

angle = π * (time_h - sunrise) / (sunset - sunrise)
irradiance = peak * sin(angle)  # zero outside daylight hours

Grid frequency and voltage models add Gaussian noise and slow drift on top of nominal values, with event injection for voltage sags and frequency deviations. All models accept a seed parameter for full determinism.

The Protocol Layer

This is where DER Twin becomes genuinely useful for integration testing — it speaks the same protocol your real devices use.

Modbus TCP

{
  "kind": "modbus_tcp",
  "ip": "0.0.0.0",
  "port": 55001,
  "unit_id": 1,
  "register_map": "bess_modbus.yaml"
}

Each device gets its own async TCP server. Multiple devices run on separate ports simultaneously. Your EMS connects exactly as it would to real hardware.

Modbus RTU

As of the latest release, DER Twin also supports Modbus RTU over serial:

{
  "kind": "modbus_rtu",
  "port": "/tmp/dertwin_device",
  "baudrate": 9600,
  "parity": "N",
  "stopbits": 1,
  "unit_id": 1,
  "register_map": "bess_modbus.yaml"
}

For development and CI without real serial hardware, use socat to create a virtual serial port pair:

socat -d -d pty,raw,echo=0,link=/tmp/sim_port pty,raw,echo=0,link=/tmp/client_port &

Point the simulator at /tmp/sim_port and your EMS client at /tmp/client_port. The rest works identically to TCP.

A single device can expose both protocols at the same time — just list both in its protocols array. A site can freely mix TCP and RTU devices.

Protocol Abstraction

The controller layer is transport-agnostic. DeviceController accesses .context and .unit_id on any protocol object regardless of whether it's TCP or RTU. Adding a new protocol means implementing two methods — run_server() and shutdown() — and registering the new kind in SiteController._build_protocol(). The device layer doesn't change at all.

IEC 61850, DNP3, and MQTT are all on the roadmap. The architecture is ready for them.

Register Maps

Register maps are YAML files that define the bridge between the protocol layer and the device layer. Each entry maps a Modbus address to a device telemetry or command field:

- name: active_power
  internal_name: active_power
  address: 32064
  func: 0x04
  direction: read
  type: int32
  count: 2
  scale: 0.1
  unit: kW

- name: on_grid_power_setpoint
  internal_name: active_power_setpoint
  address: 10126
  func: 0x10
  direction: write
  type: int32
  count: 2
  scale: 0.1
  unit: kW

name is what your EMS sees on the wire. internal_name maps to the device's internal attribute. They can differ — on_grid_power_setpoint on the wire becomes active_power_setpoint inside the BESS. This lets the simulator match real device register naming while keeping internal code clean.

Register maps are user-owned files. If you're integrating a specific manufacturer's device, you write a register map that matches their documentation and point the simulator at it. No code changes needed.

Contributing

The architecture is designed to make extensions straightforward. Full guidelines are in CONTRIBUTING.md, but the short version:

Adding a new device type:

Create dertwin/devices/<your_device>/simulator.py implementing update(dt) and get_telemetry()
Create a telemetry dataclass in dertwin/telemetry/
Add a register map YAML in configs/register_maps/
Wire it into SiteController._create_device()
Add tests

Adding a new protocol:

Implement run_server() and shutdown() in dertwin/protocol/
Register the new kind in SiteController
The device layer needs no changes

The separation between layers means you rarely need to touch more than one part of the codebase for any given change.

Documentation and Installation

Full documentation, setup guides, and API reference are at dertwin.com.

pip install dertwin

Source: github.com/AlexSpivak/dertwin

20 Minutes of Battery Operation in 0.30 Seconds

Oleksandr — Sat, 21 Mar 2026 02:38:38 +0000

That's the output of a pytest test running against a simulated 10 kWh battery — 20 minutes of physics, 0.30 seconds of wall clock time.

In Part 1 I explained why I built DER Twin — a Modbus simulator for energy devices. The short version: I spent a year building EMS software without a proper test environment and it was painful. Testing required physical hardware, every cycle took too long, and we were essentially testing on production.

This article is about what becomes possible once you have a simulator. Specifically: how to write automated tests for EMS control logic that run in seconds, reproduce any scenario deterministically, and require no hardware at all.

The Setup

To make this concrete I put together a small demo project alongside the simulator. It has a simple EMS — SimpleEMS — that connects to a BESS over Modbus TCP and cycles it between 40% and 60% SOC. There's a thin Modbus client that reads and writes registers by name, a main.py to run it against a live simulator, and the integration tests we'll walk through below.

SimpleEMS is deliberately minimal — about 60 lines. It reads SOC and working status on each poll, enables the device if it's idle, and charges or discharges depending on which side of the 40–60% window the SOC is on. Simple enough to understand immediately, realistic enough to demonstrate the testing patterns that matter.

Running the Example

DER Twin is available on PyPI so there's no need to clone the simulator itself — just install it:

git clone https://github.com/AlexSpivak/ems-demo.git
cd ems-demo
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

pip install dertwin pulls in the simulator as a dependency. No hardware, no Docker, no infrastructure.

To run the EMS against a live simulator, start the simulator in one terminal and the EMS in another:

# Terminal 1
dertwin -c site_config.json

# Terminal 2
python main.py

[EMS] Connected to BESS
[EMS] Starting in CHARGE mode
[EMS] STATUS=1.0 | SOC= 50.10% | P= -20.00 kW | MODE=charge
[EMS] STATUS=1.0 | SOC= 50.30% | P= -20.00 kW | MODE=charge

This is the manual workflow — useful for development. But the point of this article is the automated test workflow where you don't need a running simulator at all.

The Key Concept: Headless Mode

DER Twin has two modes:

real_time: true — the engine runs its own async loop. Use this when running the simulator as a standalone process that your EMS connects to over Modbus TCP.

real_time: false — the engine has no clock. You drive it step by step from your test code by calling step_once(). This is the mode that makes everything below possible.

async def run_steps(site, n: int):
    for _ in range(n):
        await site.engine.step_once()

Each call advances the simulation by one step (0.1s by default). You control time completely — you can simulate hours in milliseconds.

Building a Site in Code

SiteController is just a Python class. You can instantiate it directly without a config file — no JSON, no filesystem — which is exactly what you want in tests:

from dertwin.controllers.site_controller import SiteController

site = SiteController({
    "site_name": "test-site",
    "step": 0.1,
    "real_time": False,
    "register_map_root": "register_maps",
    "assets": [{
        "type": "bess",
        "capacity_kwh": 10.0,
        "initial_soc": 40.0,
        "max_charge_kw": 20.0,
        "max_discharge_kw": 20.0,
        "protocols": [{
            "kind": "modbus_tcp",
            "ip": "127.0.0.1",
            "port": 55501,
            "unit_id": 1,
            "register_map": "bess_modbus.yaml",
        }],
    }],
})

site.build()

Want a dual-BESS site? Add another asset to the list. Want PV and a meter? Add those too. The full site topology is a Python dict — you compose it however your test needs it.

Example 1 — Fast Forward

A 10 kWh battery charging at 20 kW takes 20 minutes to go from 40% to ~90% SOC in real life. In headless mode the same scenario runs in a fraction of a second:

@pytest.mark.asyncio
async def test_fast_forward():
    site = make_site([bess_asset(port=55501, initial_soc=40.0)])
    site.build()
    task = asyncio.create_task(site.start())

    await wait_ready(55501)

    bess = site.controllers[0].device
    bess.apply_commands({"start_stop_standby": 1, "active_power_setpoint": -20})

    # 20 minutes = 1200 seconds = 12000 steps at 0.1s
    start = time.perf_counter()
    await run_steps(site, 12000)
    elapsed = time.perf_counter() - start

    print(f"Simulated 20 minutes in {elapsed:.2f}s")
    assert bess.soc >= 89.0

Output:

Simulated 20 minutes in 0.30s
PASSED

20 minutes of battery operation in 0.30 seconds. You can simulate an entire day of operation — charge cycles, peak shaving, frequency response events — in a few seconds as part of a normal CI run.

Example 2 — Deterministic Replay

Every simulation run with the same config produces identical results. You can reproduce any scenario exactly:

@pytest.mark.asyncio
async def test_deterministic_replay():

    async def run_scenario(port: int) -> list[float]:
        site = make_site([bess_asset(port=port, initial_soc=50.0)])
        site.build()
        task = asyncio.create_task(site.start())
        samples = []
        await wait_ready(port)
        bess = site.controllers[0].device
        bess.apply_commands({"start_stop_standby": 1, "active_power_setpoint": -20})
        for _ in range(200):
            await run_steps(site, 5)
            samples.append(round(bess.soc, 4))
        await site.stop()
        task.cancel()
        return samples

    run1 = await run_scenario(port=55601)
    run2 = await run_scenario(port=55602)

    assert run1 == run2

This matters more than it might seem. If your asset is being certified for grid frequency response — FCR-D on the Nordic market, for example — you need to rehearse the exact disturbance scenario repeatedly until you're confident every register responds correctly. With a deterministic simulator you can run the same hour-long prequalification test in seconds, as many times as you need, before touching real hardware. If something fails, you can reproduce it exactly.

Example 3 — Assert on EMS Control Logic

This is the most useful pattern. Instead of running the full EMS loop as an async task — which introduces timing dependencies — we simulate the control logic directly against the device and assert it makes the right decisions at every step:

@pytest.mark.asyncio
async def test_ems_soc_bounds():
    site = make_site([bess_asset(port=55701, initial_soc=50.0)])
    site.build()
    task = asyncio.create_task(site.start())

    await wait_ready(55701)
    bess = site.controllers[0].device
    bess.apply_commands({"start_stop_standby": 1})
    await run_steps(site, 10)

    soc_samples = []
    mode = "charge"

    for _ in range(8000):
        await run_steps(site, 1)
        soc = bess.soc
        soc_samples.append(soc)

        if mode == "charge":
            bess.apply_commands({"active_power_setpoint": -20})
            if soc >= 60.0:
                mode = "discharge"
        elif mode == "discharge":
            bess.apply_commands({"active_power_setpoint": 20})
            if soc <= 40.0:
                mode = "charge"

    print(f"SOC range: {min(soc_samples):.1f}% – {max(soc_samples):.1f}%")

    assert min(soc_samples) >= 38.0
    assert max(soc_samples) <= 62.0

Output:

SOC range over 8000 steps: 40.0% – 60.0%
PASSED

8000 steps is 800 simulated seconds — roughly 13 minutes of BESS operation, enough for several full charge/discharge cycles. The control logic from SimpleEMS is mirrored directly in the test loop, which means we're testing the exact same decision logic the EMS runs in production. On real hardware this would take actual minutes per cycle with no way to assert on intermediate states.

All Three Tests Together

pytest test_examples.py -v

test_examples.py::test_fast_forward PASSED
test_examples.py::test_deterministic_replay PASSED
test_examples.py::test_ems_soc_bounds PASSED

3 passed in 1.71s

Three tests covering fast-forward, determinism, and control logic correctness. 1.71 seconds total. The equivalent real-hardware test time would be measured in hours.

What This Enables

Once you have this pattern in place you can write tests for things that were previously impossible to automate:

SOC boundary violations — does your EMS ever accidentally overcharge or overdischarge?
Ramp rate behavior — does power ramp correctly when you change setpoints?
Mode transition logic — does the EMS switch modes at exactly the right thresholds?
Recovery after fault — what happens when the BESS trips and comes back online?
Multi-asset coordination — if you have two BESS units, do they stay independent?

All of these become standard pytest tests that run in CI on every push. No hardware. No lab. No waiting.

The Code

Demo project with all examples:
github.com/AlexSpivak/ems-demo

Simulator library:
github.com/AlexSpivak/dertwin — pip install dertwin

In Part 3 I'll walk through the simulator architecture — how the engine, device models, and protocol layer are structured, and how to add support for new device types.

Why I Built a Modbus Simulator After a Year of Testing on Production

Oleksandr — Tue, 10 Mar 2026 18:23:04 +0000

While leading engineering at an energy management system startup, I ran into a problem that kept getting worse the longer I ignored it.

Whenever my engineer and I needed to test something — even something as basic as reading telemetry from a battery — someone had to physically connect a device, pull a Docker image to the field hardware, run the code against it, and check whether the data made sense. Either by inspecting logs locally or watching values flow through the queue.

This is done even for the simplest operation like collecting telemetry. A read operation. Of course with write operation it was getting even more complicated.

We practically were testing on production. The assets weren't prequalified yet, every test cycle took longer than it should have, and I kept thinking about what happens when these batteries are actually running 24/7. You can't afford to pull a Docker image to a production BESS at 2am to check if your register parser is correct.

We were a small team moving fast and building a proper test environment kept getting pushed. So we kept going the way we had been — carefully, slowly, and with more anxiety than necessary.

The testing loop without a simulator looked pretty much like this:

What We Were Building

For context: we were building an Energy Management System from scratch. The stack was real — GKE, MQTT, TimescaleDB, Pub/Sub, BigQuery, Dagster, Grafana, Raspberry Pis in the field talking Modbus to actual batteries. Up to five engineers at peak. Real assets, real protocols, real production pressure.

The core job of an EMS is to talk to energy devices — batteries, inverters, meters — read their state, and send commands. In our case we were preparing to participate in the Swedish frequency regulation market — FCR-D up, FCR-D down, FCR-N. These are ancillary services where your battery responds to disturbances in grid frequency, charging or discharging within seconds to help stabilise the grid. The protocol between EMS and device is usually Modbus TCP: a simple binary protocol over TCP where you read and write registers at specific addresses.

The prequalification process for these markets is a formal test — roughly an hour where the TSO verifies your asset actually responds correctly to frequency disturbances. You can unit test the algebra behind your control logic all day. But at some point you have to run your code against a real device and watch every register react correctly to a simulated disturbance event. One hour, one shot, no room for a parsing bug you didn't catch.

The problem with testing this kind of system is that the other side of the connection is a physical device. You can't mock a Modbus server in a unit test and call it done — register maps differ between manufacturers, scale factors are easy to get wrong, and the only way to know your telemetry parsing is correct is to see real values come back from something that behaves like the actual hardware.

Without a simulator, every test required the physical device. And physical devices are expensive, shared, and inconveniently located.

The Idea

So over the last few weeks I built what I wished we'd had: a simulator for distributed energy resources that speaks real Modbus TCP, runs on your laptop, and behaves like actual hardware.

Instead of connecting your EMS to a real battery, you connect it to a simulator that:

Accepts Modbus TCP connections on configurable ports
Exposes the same register map as the real device
Actually models the physics — SOC changes when you charge, power ramps up at the configured rate, PV output follows an irradiance model
Lets you inject grid events — voltage sags, frequency deviations — to test fault handling
Runs deterministically so you can reproduce the same scenario twice

You define your site in a JSON config file — which devices, which ports, what parameters — and the simulator builds the whole thing. Want a dual-BESS site with a PV inverter, energy meter, and a grid frequency model starting at solar noon? That's a config file.

The project is called DER Twin — github.com/AlexSpivak/dertwin — and the rest of this article walks through what it looks like in practice.

What It Looks Like

Start the simulator:

python -m dertwin.main -c configs/simple_config.json

INFO | Building site: local-dev-site
INFO | Starting Modbus server | 0.0.0.0:55001 | unit=1
INFO | Simulation engine started | step=0.100s

Connect your EMS — or the included example — from another terminal:

python examples/main_simple.py

[EMS] Connected to BESS
[EMS] Starting in CHARGE mode
[EMS] STATUS=1 | SOC= 42.30% | P=  -20.00 kW | MODE=charge
[EMS] STATUS=1 | SOC= 44.10% | P=  -20.00 kW | MODE=charge
[EMS] Reached 60% → switching to DISCHARGE

The EMS is talking Modbus TCP to a simulated BESS. SOC is changing. Power is ramping. It works exactly like the real thing — because from the EMS's perspective, it is the real thing.

For a full site with PV producing at noon, dual BESS cycling independently, and an energy meter tracking net grid power:

[SITE] Grid= -34.63 kW | PV= 24.60 kW (producing) | Freq=50.000 Hz
  [BESS-1] RUN  | SOC= 59.8% | P= +50.00 kW | MODE=discharge
  [BESS-2] RUN  | SOC= 40.2% | P= -25.00 kW | MODE=charge

Grid exporting 34.6 kW because PV is generating 24.6 kW and BESS-1 is discharging at 50 kW while BESS-2 charges at 25 kW. The math is right. The physics is right.

Why This Changes the Development Loop

With a simulator, the workflow becomes straightforward:

Write your Modbus client code
Start the simulator
Run your code against it
If something is wrong, fix it immediately — no hardware, no waiting, no pulling Docker images at 2am
Write integration tests that run the simulator headlessly and assert on telemetry values
Merge with confidence

That last point is the one I care about most. The simulator supports a headless mode where there is no real-time clock — you drive it step by step from your test code, the same way test_site_controller.py does in the project. That means you can write integration tests that spin up a full simulated site, run thousands of simulation steps in milliseconds, and assert that your EMS made the right decisions at every point. No hardware, no timing dependencies, no flaky tests.

The fragile, anxiety-inducing "test on production" loop is gone. You know your code works before it touches a real device.

Now the testing loop with a simulator looks like this:

The Project

DER Twin is open source and MIT licensed.

In the next article I'll walk through the architecture — how the engine, device models, and protocol layer are separated, and why that separation matters if you want to add new device types or protocols.

If you're building EMS software or working with Modbus — let me know if this is useful.