Write your own source#

battfeed’s reason for existing is the sources you write: the cycler in your lab, the BMS on your bench. In this tutorial you will implement a source from scratch, prove it correct with the shipped contract test, run it, and see what it takes to publish it as a plugin. The example is deliberately trivial — a voltage ramp — so every line is about the contract, not the device.

1. The whole contract is three members#

battfeed.DataSource is a typing.Protocol — the seam is structural, so your class imports nothing from battfeed. Save as custom_source.py:

class RampSource:
    name = "ramp"

    def __init__(self) -> None:
        self._step = 0

    def metadata(self):
        return {"source": self.name, "vendor": "example", "kind": "synthetic-ramp"}

    def poll(self):
        self._step += 1
        return [
            {
                "voltage_volt": 3.0 + 0.01 * self._step,  # charging: voltage rising
                "current_ampere": 0.05,  # positive = charging (BDF sign convention)
            }
        ]

    def close(self):  # optional hook; battfeed calls it when present
        print("RampSource closed")

name identifies the source, metadata() describes it (this lands in every sidecar), and poll() returns zero or more new samples since the last call, keyed by canonical BDF column names. A real source would talk to hardware inside poll().

The one rule that matters most: when the device is unreachable, raise. Do not write a retry loop — the Harvester owns retry, with exponential backoff under a configurable ErrorPolicy, and only abandons the run after too many consecutive failures. This division of labour is what makes multi-day collection survive flaky Bluetooth, and it only works if sources stay honest about failure. The reasoning gets a full page: how battfeed survives flaky hardware.

2. Prove it correct with one line#

battfeed.testing.check_source asserts everything a pipeline relies on — the protocol surface, strict-JSON metadata and samples, sample shape, routing-key discipline. Save as test_ramp.py:

from battfeed.testing import check_source
from custom_source import RampSource

def test_ramp_contract():
    check_source(RampSource())  # drive it with a mock or replay tape, never live hardware

$ python -m pytest test_ramp.py -q
.                                                                        [100%]
1 passed in 0.11s

That one line is a complete contract test; put it in your own package’s suite and battfeed’s expectations are enforced on every commit you make.

3. Run it#

from battfeed import BdfCsvSink, Harvester

harvester = Harvester()
harvester.register(RampSource())
sink = BdfCsvSink("LOCAL__RampCell__20260811_001.bdf.csv")
stats = harvester.collect("ramp", duration_s=5, interval_s=0.5, sink=sink)
sink.close()
print(f"Wrote {stats.samples} samples; columns: {stats.columns}")

$ python custom_source.py
Wrote 10 samples; columns: ['current_ampere', 'test_time_second', 'voltage_volt']

Ten polls in five seconds, each returning one sample; the harvester stamped test_time_second because the source did not supply it.

4. Publish it as a plugin#

To make your source appear in battfeed sources and work with battfeed collect --source ramp, declare an entry point in your own package’s pyproject.toml — battfeed’s own built-ins register through this exact mechanism, so it is exercised constantly:

[project.entry-points."battfeed.sources"]
ramp = "my_pkg.sources:RampSource"

Two optional hooks make a source a better citizen: an availability() classmethod that explains why the source cannot run here (missing extra, wrong platform — this is what produces the [unavailable: ...] annotations in battfeed sources), and a discover() classmethod that lets battfeed discover find your devices and print ready-to-paste collect commands.

5. Push-style hardware#

Devices that push readings (BLE notifications, CAN frames, MQTT) don’t fit a synchronous poll(). Subclass battfeed.StreamingSource instead: implement a blocking run_reader(emit, should_stop) loop calling emit(sample) per reading, and the base class runs it in a background thread and buffers samples for poll() to drain — bounded, with overflow counted rather than silently dropped, and a dead reader re-raises at the next poll() so the harvester’s ErrorPolicy still owns retry.

Because BLE/CAN hardware is not on every desk, the supported workflow is replay-first: record raw frames from one live session into a JSONL tape (battfeed.testing.TapeRecorder), commit the tape, and drive every test from it with ReplayReader — which compresses time, so an hour-long session replays in milliseconds. See examples/streaming_source.py for a complete runnable version.

6. Where you are now#

You have implemented, verified, run, and (on paper) published a battfeed source. The API reference documents every object you touched; design principles explains why the seam is structural and the core dependency-free.