API reference#

The public API is importable from the top-level battfeed package; the modules below are where the objects live. Anything not documented here (names starting with _) is internal.

battfeed.protocols: the contracts#

Public contracts for battfeed data sources and sinks.

This module is the stable seam of battfeed: third-party collectors implement DataSource, output writers implement Sink, and everything else in the package is wiring between the two.

Both contracts use typing.Protocol (structural typing), so an implementation never needs to import or subclass anything from battfeed – any object with the right attributes and methods satisfies the contract. This is deliberate: commercial platforms can ship proprietary sources and sinks that plug into battfeed without depending on its internals.

Sample shape#

A sample is a plain dict whose keys are canonical machine-readable BDF (Battery Data Format) column names of the form {quantity}_{unit}, for example:

{"test_time_second": 12.0, "voltage_volt": 3.71, "current_ampere": -0.002}

Values are normally numeric. String values are permitted for auxiliary columns (e.g. a charge status flag) – they are written to the CSV as-is, but note that strict BDF validation flags columns outside the canonical vocabulary. Slow-changing facts (device model, chemistry, firmware) belong in DataSource.metadata(), not in every sample.

Sign convention (per the Battery Data Format specification): positive current charges the test object, negative current discharges it.

Reserved routing keys#

Two keys in RESERVED_KEYS are part of the sample contract but are not battery measurements and are never BDF columns – they are routing metadata, stripped before any BDF output:

  • series_id – which physical object a sample belongs to (which car, which pack, which bay). One connection can yield many objects.

  • run_id – which test/run segment the sample belongs to. One object can yield many runs; without run_id the readings of several runs would merge into a single non-monotonic timebase, which is invalid BDF.

Both are optional per sample, both are str when present, and both are stripped by every sink before writing (see battfeed.BdfCsvSink), so a routing-aware source wired to a plain, non-routing sink can never leak them into CSV columns. A RoutingSink (a later work package) uses them to demultiplex one stream into one BDF file per (series_id, run_id).

Timebase ownership (invariant I5). A source that emits these routing keys must supply its own test_time_second, zero-based per (series_id, run_id). The harvester’s fallback stamp is a shared elapsed-collection time measured from the start of the run; that is only correct for single-object sources. An object that comes online two hours into a collection must start its file at t = 0, not t = 7200 – so a routing source owns its own per-(series, run) clock rather than relying on the harvester.

class battfeed.protocols.DataSource(*args, **kwargs)[source]#

A live source of battery samples.

This protocol is the stable seam that third-party collectors implement. Implementations are structural: define name, metadata() and poll() on any class and it is a DataSource – no import or inheritance required. Register it with a battfeed.Harvester directly, or expose it to the battfeed CLI through the "battfeed.sources" entry-point group.

Error handling#

poll() MAY raise when the underlying device is briefly unreachable; the harvester’s battfeed.ErrorPolicy retries with backoff, so sources should NOT implement their own retry loops. A source should only swallow errors it can genuinely resolve better itself (e.g. one bad frame out of several channels).

Optional hooks#

close() -> None releases hardware handles, serial ports, network sessions, and so on; callers invoke it when present. A classmethod availability() -> str | None may report why the source cannot run here (missing optional dependency, wrong platform); the CLI uses it to annotate battfeed sources. A classmethod discover(timeout_s=..., **options) -> list[dict] may scan for connectable devices; each candidate dict carries "option" and "value" (the constructor kwarg the value plugs into, e.g. address/serial), optionally "ready": False when a candidate needs a manual step before it can be collected from, plus free-form descriptive fields. The CLI surfaces this as battfeed discover, and sources may accept the same "auto" sentinel for the selector kwarg to self-resolve an unambiguous device. None of these are part of the required protocol, so trivial sources stay trivial.

name: str#

Short unique identifier for the source, e.g. "simulator".

metadata()[source]#

Return a static description of the source.

Called at most a handful of times per collection run (never in the hot loop). The mapping should be JSON-serialisable; it is recorded in the .meta.json sidecar written next to each BDF file.

Return type:

Mapping[str, Any]

poll()[source]#

Return zero or more NEW samples accumulated since the last call.

Keys must be canonical BDF column names (voltage_volt, current_ampere, …). A source MAY include test_time_second itself (e.g. when tailing an instrument log that records its own timebase); when absent, the harvester stamps each sample with the elapsed collection time.

A sample MAY also carry the routing keys in RESERVED_KEYS (series_id / run_id); they route the sample to a per-object, per-run BDF file and are never written as columns. A source that emits them must supply its own zero-based-per-(series, run) test_time_second (invariant I5) – the harvester’s shared elapsed-collection stamp is wrong for objects that appear mid-run. See the module docstring for the full routing contract.

Must not block for longer than roughly one polling interval and must never return the same sample twice.

Return type:

list[dict[str, float | int | str]]

battfeed.protocols.RESERVED_KEYS: tuple[str, ...] = ('series_id', 'run_id')#

Sample keys that route rather than measure; stripped before BDF output.

See the module docstring for the routing contract. series_id identifies which physical object a sample belongs to; run_id identifies which test-run segment. Both are optional and str when present, and no sink ever writes them as CSV columns.

class battfeed.protocols.Sink(*args, **kwargs)[source]#

A destination for collected samples.

This protocol is the stable seam that output writers implement – the harvester only ever calls write() and the owner of the sink calls close() exactly once when collection is finished. battfeed ships battfeed.BdfCsvSink, which writes BDF CSV files; alternative sinks (message queues, databases, platform ingest APIs) just need these two methods.

write(rows)[source]#

Persist a batch of samples. May be called with an empty batch.

Parameters:

rows (Iterable[Mapping[str, float | int | str]])

Return type:

None

close()[source]#

Flush and release resources. Must be idempotent.

Return type:

None

battfeed.protocols.Sample#

canonical BDF column names mapped to values.

Type:

One reading

alias of dict[str, float | int | str]

battfeed.protocols.SampleValue#

A single measured value; numeric for canonical BDF columns.

alias of float | int | str

battfeed.harvester: the collection loop#

Polling loop that moves samples from a DataSource into a Sink.

The Harvester owns no I/O of its own: sources produce samples, sinks persist them, and the harvester just runs the clock. The clock and sleep functions are injectable so the loop can be tested (and simulated) without waiting on wall time.

Resilience: field collection has to survive flaky hardware. A transient poll() failure does not abort the run – the harvester retries with exponential backoff under a configurable ErrorPolicy and only gives up (raising SourceFailure) after too many consecutive failures. Sources therefore stay simple: raise on trouble, reconnect on the next poll.

class battfeed.harvester.CollectStats(samples, duration_s, started_at, source, columns=<factory>, errors=0)[source]#

Summary of one Harvester.collect() run.

Parameters:
samples: int#

Total number of samples written to the sink.

duration_s: float#

Elapsed collection time in seconds (measured with the injected clock).

started_at: str#

Wall-clock start of the run as an ISO 8601 UTC timestamp.

source: str#

Name of the source that was collected.

columns: list[str]#

Sorted union of the column names seen across all samples.

errors: int = 0#

Number of tolerated (retried) poll failures during the run.

class battfeed.harvester.ErrorPolicy(max_consecutive_errors=5, backoff_initial_s=1.0, backoff_factor=2.0, backoff_max_s=30.0)[source]#

How Harvester.collect() treats poll() failures.

A failed poll is logged and retried after an exponentially growing delay; any successful poll resets the consecutive-failure counter. Once max_consecutive_errors failures occur in a row the run is abandoned with SourceFailure.

Parameters:

  • max_consecutive_errors (int)

  • backoff_initial_s (float)

  • backoff_factor (float)

  • backoff_max_s (float)

backoff(consecutive)[source]#

Delay before the next attempt after consecutive (>=1) failures.

Parameters:

consecutive (int)

Return type:

float

class battfeed.harvester.Harvester[source]#

Registers named sources and runs timed collection loops against them.

Typical use:

harvester = Harvester()
harvester.register(SimulatedCellSource())
sink = BdfCsvSink("LOCAL__DemoCell__20260707_001.bdf.csv")
stats = harvester.collect("simulator", duration_s=60, interval_s=1.0, sink=sink)
sink.close()

The harvester never closes the sink and never closes the source: their owner (your script, or the battfeed CLI) does. This keeps repeated collections against the same source or sink possible.

register(source)[source]#

Register source under its name. Re-registering replaces it.

Parameters:

source (DataSource)

Return type:

None

property sources: dict[str, DataSource]#

Mapping of registered source names to source objects (a copy).

status(source_name)[source]#

Return a status snapshot for source_name.

The dict always contains registered (bool), last_poll_at (ISO 8601 string or None) and samples_collected (int, total across all collect runs in this harvester’s lifetime).

Parameters:

source_name (str)

Return type:

dict[str, Any]

collect(source_name, *, duration_s=None, interval_s=1.0, sink, errors=ErrorPolicy(max_consecutive_errors=5, backoff_initial_s=1.0, backoff_factor=2.0, backoff_max_s=30.0), clock=<built-in function monotonic>, sleep=<built-in function sleep>, stop=None)[source]#

Poll source_name every interval_s seconds into sink.

With duration_s=None (the default) the loop runs until stop is set – the mode field collectors use; pass a number of seconds for a bounded run. Each iteration: poll the source, stamp test_time_second (elapsed time from the injected clock) onto samples that lack it, hand the batch to sink.write, then sleep until the next tick.

poll() exceptions are governed by errors: transient failures are logged and retried with exponential backoff, and only errors.max_consecutive_errors failures in a row abandon the run with SourceFailure. Pass errors=None to fail fast on the first exception instead.

clock and sleep exist so tests can drive the loop with fake time; production callers keep the defaults.

Raises:

  • KeyError – if source_name is not registered.

  • ValueError – if interval_s is not positive, or neither duration_s nor stop is provided (which would loop forever with no way to end).

  • SourceFailure – if the source exceeds its error allowance.

Parameters:
Return type:

CollectStats

exception battfeed.harvester.SourceFailure(source, consecutive, last_error)[source]#

A source kept failing beyond its ErrorPolicy allowance.

Parameters:
Return type:

None

battfeed.registry: source discovery#

Discovery of installed data sources.

Sources are found in two places:

  1. The built-in sources shipped with battfeed (simulator, csvtail, wmi).

  2. The "battfeed.sources" entry-point group, which any installed package can contribute to:

    [project.entry-points."battfeed.sources"]
my-cycler = "my_pkg.sources:MyCyclerSource"

battfeed registers its own built-ins through the same entry-point group (see pyproject.toml), so the plugin mechanism is exercised on every install; the built-in table below only guarantees discovery when battfeed is imported from a source tree without being installed.

battfeed.registry.available_sources()[source]#

Return every discoverable source class, keyed by source name.

Built-ins are always present; entry points contribute additional names. A broken source is skipped with a warning rather than breaking discovery for everyone else.

Return type:

dict[str, type]

battfeed.registry.create_source(name, **kwargs)[source]#

Instantiate the source registered under name.

Keyword arguments are passed through to the source constructor, e.g. create_source("csvtail", path="log.csv", column_map={...}).

Raises:

KeyError – if no source is registered under name.

Parameters:

name (str)

battfeed.config: config files and secrets#

Run configuration from a TOML file, plus the credential-hygiene helpers.

Long battfeed invocations carry a lot of state: an mc3000 slot and BLE address, a dji folder path and keychain key, a push endpoint URL and bearer token. Passing all of that as --opt KEY=VALUE flags is unwieldy and, worse, leaks secrets into shell history and the process list (ps / Task Manager show a process’s full argv to any local user). This module lets those settings live in a file instead, and it centralises the redaction that keeps secrets out of logs, error messages, the battfeed sources listing, and the .meta.json sidecars.

Redaction is two complementary layers, because neither alone is enough:

  • By key name (is_secret_key() / redact_mapping()) – a value whose key looks like a credential (*key* / *token* / *secret* / *password* / *passphrase* / *credential* / *authorization* / *bearer* / *session* / *cookie* / *signature* / *private*, and the short whole-segment auth / pat / pin / otp / salt) is masked wherever it appears. The name is split into snake/kebab/camel segments so pat masks pat but not path.

  • By value + URL (redact_text() / mask_url_userinfo()) – the concrete secret values resolved for a run are scrubbed from any text, and scheme://user:pass@host userinfo is masked, so a credential hiding under an innocuous key (endpoint as a URL, a duplicated token under note) is caught even though its key name is not suspicious.

Honest residuals (not claims of perfection): a secret a source reads from its own env var without going through a registered (kwarg, env) mapping is known only to that source (it must self-redact); a malformed-TOML parse error can echo an inline secret before any value is resolved (so prefer ${ENV:VAR}, which keeps the secret out of the file entirely); and a handler a source both creates and logs to entirely within its own __init__ is masked only if it existed before construction.

Config file shape (TOML)#

[collect]                      # defaults for `battfeed collect`
interval = 2.0
institution = "SINTEF"
cell = "Pack-07"

[import]                       # defaults for `battfeed import`
interval = 5.0
out_dir = "imported"
watch = true

[source.mc3000]                # options for `--source mc3000`
slot = 1
transport = "ble"
address = "AA:BB:CC:DD:EE:FF"

[source.dji]                   # options for `--source dji`
type = "dji"                   # optional: the registered source *type*;
path = "C:/logs/dji"           #   defaults to the block name ("dji" here),
api_key = "${ENV:DJI_API_KEY}" #   so a block may alias a source under a
                               #   different name, e.g. [source.bay2].

A [source.<name>] block matches --source <name>. Its type field (if absent, the block name) selects the registered source class; every other key is a constructor keyword argument.

Precedence (explicit, and pinned by tests)#

For a source option (a source constructor kwarg):

--opt KEY=VALUE   >   [source.<name>] in the config   >   the source's own
                                                          default

For a run parameter (interval / institution / cell / duration / out / out_dir / watch):

the CLI flag   >   [collect] or [import] in the config   >   the built-in
                                                             default

There is no dedicated CLI flag for individual source kwargs – --opt is that layer – so --opt overriding the config file is the source-kwarg spelling of “the CLI wins”. The environment enters through ${ENV:VAR} expansion (below), which is the sanctioned way to keep a secret out of the file itself; a few sources also read their own env var (e.g. DJI_API_KEY) as a last resort when the kwarg is left unset, which sits at the “default” layer.

Environment expansion#

Any string value containing ${ENV:VAR_NAME} is replaced with the value of that environment variable at load time. A reference to an unset variable is a hard error (better than silently sending an empty key on the wire). Write api_key = "${ENV:DJI_API_KEY}" and the secret never touches the file.

Python 3.10#

Parsing TOML uses tomllib, which is standard library on Python 3.11+. battfeed supports 3.10, where tomllib does not exist, so the import is conditional and falls back to the optional tomli backport if it happens to be installed (no hard dependency is added to the core – invariant I6). On 3.10 without tomli, --config raises an actionable error pointing at the upgrade or the one-line pip install tomli; every other feature works unchanged.

battfeed.config.REDACTED = '***'#

The mask substituted for any secret value that would otherwise be echoed.

class battfeed.config.Config(collect=<factory>, import_=<factory>, sources=<factory>)[source]#

A parsed battfeed config file; ${ENV:VAR} is expanded lazily on access.

collect / import_ hold the raw run-parameter tables; sources maps each [source.<name>] block name to its raw {type?, **kwargs} dict. Expansion happens only in run_params() / source_block(), when a section is actually consumed – so an unset ${ENV:VAR} in a block the run does not use (a different source, or the other verb’s section) never derails the run. Each accessor returns a fresh copy.

Parameters:
run_params(verb)[source]#

Return the env-expanded run-parameter table for collect/import.

Parameters:

verb (str)

Return type:

dict[str, Any]

source_block(name)[source]#

Return the env-expanded [source.<name>] block, or None.

Parameters:

name (str)

Return type:

dict[str, Any] | None

exception battfeed.config.ConfigError[source]#

A config file is missing, unparseable, or structurally invalid.

Subclasses ValueError so the CLI’s existing actionable-error handling surfaces the message without a traceback.

battfeed.config.is_secret_key(name)[source]#

True if an option/metadata key name looks like it holds a secret.

The name is split into word segments (on _ / - / digits / camelCase); a segment matches if it equals one of the short whole-segment secrets (auth/pass/cred/pat/pin/otp/sig/salt) or contains one of the longer unambiguous ones (key/token/secret/ password/passphrase/credential/authorization/signature/ session/cookie/private/bearer and kin). This masks authorization / bearer_token / db_password / private_key / session_cookie / passphrase while leaving path / format / transport / compatibility / interval / institution alone – the segment split is what keeps pat off path and salt off asphalt.

Parameters:

name (str)

Return type:

bool

battfeed.config.load_config(path)[source]#

Load and validate the TOML config at path (env expansion is lazy).

${ENV:VAR} references are resolved when a section is read via Config.run_params() / Config.source_block(), not here, so an unset variable in an unused block does not break an unrelated run.

Raises:

ConfigError – if the file is missing or unreadable, is not valid TOML, has a malformed [collect] / [import] / [source.*] section, or is given while no TOML parser is available (Python 3.10 without tomli). An unset ${ENV:VAR} is reported later, when the section that references it is used. Every message is actionable on its own – the CLI prints it without a traceback.

Parameters:

path (str | Path)

Return type:

Config

battfeed.config.mask_url_userinfo(text)[source]#

Mask user:pass@ userinfo in every scheme://user:pass@host in text.

https://u:SECRET@host/x -> https://***@host/x. Generalises HttpPushSink’s per-URL redaction to free text (a log line, an error, a metadata value) where a credential may hide under a non-secret key name.

Parameters:

text (str)

Return type:

str

battfeed.config.redact_mapping(mapping, secrets=())[source]#

Deep-copy mapping with secrets masked by BOTH key name and value.

Key-name layer: any value whose key is_secret_key() becomes ***. Value layer (applied to every remaining string, at any depth via redact_text()): known secret values in secrets are scrubbed and URL userinfo is masked – so a credential hiding under an innocuous key (an endpoint URL, a copy of the token under note) is caught too. Recurses through nested mappings, lists and tuples; the input is never mutated. This is the pass used at the .meta.json sidecar boundary.

Parameters:
Return type:

dict[str, Any]

battfeed.config.redact_text(text, secrets=())[source]#

Scrub known secret values from text and mask any URL userinfo.

Two value-based layers that complement the key-name masking of redact_mapping(): every literal in secrets (longest first, so a secret that contains another is fully masked) becomes ***, and any scheme://user:pass@host credential is masked even when its value was never a named option. Safe on empty inputs.

Parameters:
Return type:

str

battfeed.config.redact_value(name, value)[source]#

Return value unless name is a secret key, then REDACTED.

Parameters:
Return type:

Any

battfeed.config.url_userinfo_passwords(text)[source]#

Return the password component of every scheme://user:pass@ in text.

Harvested into the run’s secret-value set so a credential embedded in a URL under an innocuous key (endpoint, dsn) is scrubbed by value from logs and errors too, not only masked in place.

Parameters:

text (str)

Return type:

list[str]

battfeed.importer and battfeed.ingest_state: file import#

Batch-import driver: poll a file-ingesting source to exhaustion, or forever.

Importers are ordinary DataSources (invariant I3 – one seam, no parallel pipeline): a folder-watching source parses the next un-ingested file per poll() and returns its rows as one batch. What a batch import needs that live collection does not is a different loop: not “poll every N seconds for a duration” but “poll until there is nothing left” (one-shot) or “poll forever, idling between checks for new files” (watch). run_import() is that loop – a thin driver over the same poll() contract, nothing more.

Drained detection#

One-shot mode stops when the source reports drained. A source MAY implement the optional hook drained() -> bool; it is consulted after every empty poll, so a source that knows more files are pending (e.g. it quarantined one and wants another look, or ingestion is deliberately paced) can return False to keep the driver polling. A source without the hook is considered drained after its first empty poll. Because a drained() hook can keep a one-shot run polling indefinitely, one-shot mode on a hook-bearing source requires a stop event, exactly like watch mode (the CLI always passes one).

Commit point and at-least-once delivery#

A batch source keeping durable dedupe state (see battfeed.ingest_state.ImportLedger) must NOT mark a file as ingested inside poll() – at that moment its rows exist only in memory, and a crash or sink failure before the write completes would lose the file forever while the ledger swears it was imported. The commit point lives after the write: when sink.write(batch) returns successfully, the driver calls the optional source hook commit_batch() -> None, and THAT is where the source records the batch’s file in its ledger. The resulting semantics are at-least-once: a crash between poll and commit leaves the file un-recorded, so the next run imports it again. Duplicates from such a re-run land in NEW segment files (never overwriting the earlier ones) because RoutingSink reserves every output path atomically – re-imported data is a visible, de-duplicable artifact, not silent corruption. The previous design (record inside poll()) was silently at-most-once: a crash in the poll-to-write window lost the file permanently.

Error handling#

Sources raise on trouble; the driver owns retry (invariant I4). Rather than duplicating backoff logic, the driver reuses the harvester’s ErrorPolicy – the same consecutive-failure accounting and exponential backoff, abandoning the run with SourceFailure only after max_consecutive_errors failures in a row.

Timebase ownership (invariant I5)#

Unlike Harvester.collect(), this driver never stamps test_time_second: imported files carry their own timebase and their rows carry routing keys (imported data is inherently multi-(series, run)), so the source must supply test_time_second zero-based per (series, run). Rows are passed to the sink exactly as the source produced them.

The driver never closes the source or the sink: their owner (your script, or the battfeed import CLI verb) does – mirroring the harvester.

CLI hook: reset_ledger()#

Batch sources keep durable dedupe state (see battfeed.ingest_state.ImportLedger) at a location of their own choosing, so the CLI cannot clear it directly. A source that keeps a ledger SHOULD expose an optional reset_ledger() -> None hook; battfeed import --reset-ledger calls it (and fails loudly on sources that lack it).

class battfeed.importer.ImportStats(samples, polls, batches, duration_s, started_at, source, columns=<factory>, errors=0)[source]#

Summary of one run_import() run.

Parameters:
samples: int#

Total number of samples written to the sink.

polls: int#

Number of successful poll() calls (failed polls are errors).

batches: int#

Number of non-empty batches written to the sink.

duration_s: float#

Elapsed driver time in seconds (measured with the injected clock).

started_at: str#

Wall-clock start of the run as an ISO 8601 UTC timestamp.

source: str#

Name of the source that was drained/watched.

columns: list[str]#

Sorted union of the column names seen across all samples.

errors: int = 0#

Number of tolerated (retried) poll failures during the run.

battfeed.importer.run_import(source, sink, *, watch=False, interval_s=5.0, stop=None, errors=ErrorPolicy(max_consecutive_errors=5, backoff_initial_s=1.0, backoff_factor=2.0, backoff_max_s=30.0), clock=<built-in function monotonic>, sleep=None)[source]#

Poll source and write its batches to sink until drained (or stopped).

Each successful non-empty poll is written to the sink and followed immediately by the next poll – a source that keeps returning rows is drained at full speed, with no sleeps at all. Only an empty poll idles: in watch mode the driver waits interval_s and polls again, forever, until stop is set; in one-shot mode (the default) an empty poll ends the run once the source reports drained (see the module docstring for the optional drained() hook).

After each successful sink.write(batch) the driver calls the source’s optional commit_batch() hook – the commit point where a batch source records the ingested file in its ledger (see the module docstring: at-least-once, never silently at-most-once). A sink.write failure propagates immediately (it is not retried by errors, which governs poll() only) and the un-committed batch is re-imported on the next run.

poll() exceptions are governed by errors exactly as in Harvester.collect(): transient failures are logged and retried with exponential backoff, and only errors.max_consecutive_errors failures in a row abandon the run with SourceFailure. Pass errors=None to fail fast on the first exception instead.

stop is honoured in both modes: it is checked between polls, and when sleep is left at its default every wait (idle interval and error backoff alike) is stop.wait, so setting the event interrupts even a long backoff immediately. Pass clock and sleep to drive the loop without wall time in tests (an explicit sleep is used verbatim and is then responsible for its own stop-responsiveness).

Rows are BDF-validity-checked only for the one thing the driver can see: a batch source owns its per-(series, run) timebase (invariant I5), so the first batch containing rows without test_time_second draws a single warning per run naming the source.

Raises:

  • ValueError – if interval_s is not positive, or watch=True without a stop event, or one-shot mode on a source with a drained() hook without a stop event (either could loop forever with no way to end).

  • SourceFailure – if the source exceeds its error allowance.

Parameters:
Return type:

ImportStats

Durable import state for batch sources: a dedupe ledger and a quarantine.

A folder-watching import source must answer two questions across process restarts: have I ingested this file before? and is this file known to be permanently unsupported? ImportLedger answers both from a single JSON file.

Content hashes, not paths#

Both the ingested set and the quarantine are keyed by the sha256 of the file’s content, not its path: a re-plugged SD card that mounts under a new drive letter must not re-ingest, and a corrupt .DAT copied somewhere else is remembered, not retried. The path and a timestamp are recorded alongside each hash purely as human-readable provenance. The one exception is zero-byte files: every empty file shares one hash, so an entry for it would make all empty files “seen” (or worse, inherit each other’s quarantine reasons). Empty files are therefore never content-keyed – seen() and is_quarantined() always answer False for them, record() and quarantine() decline (at info level), and each empty file is judged on its own every run (it may simply still be being written).

Commit point: record only AFTER the write#

Do NOT call record() inside poll(). At that moment the file’s rows exist only in memory; a crash or sink failure before they reach the sink would lose the file forever while the ledger swears it was imported (silent at-most-once). Defer record() into the source’s commit_batch() hook, which battfeed.run_import() calls only after sink.write returned successfully – the semantics become at-least-once: a crash between poll and commit leaves the file un-recorded and it is re-imported on the next run, into NEW segment files (the routing sink reserves output paths atomically, so a re-import never overwrites earlier data).

Quarantine vs. dedupe#

The two sets answer different questions. record() marks a file as successfully ingested; quarantine() marks it as permanently unsupported, with a reason the operator can read back (no silent data loss – invariant I2: a skipped file is counted and explains itself). A transient parse failure belongs in neither: raise from poll() and let the driver’s ErrorPolicy retry (invariant I4).

ONE writer at a time#

A ledger file supports a single writer at a time. Running concurrent imports over one ledger is unsupported. Every mutation rewrites the whole file from this instance’s in-memory state, so two concurrent writers silently lose each other’s updates (last writer wins) – and a lost entry is a file that will be re-imported. The atomic unique-temp-file writes below guarantee the file is never corrupted by concurrent writers or crashes, not that their updates merge. Cross-process locking is deliberately out of scope for now.

Ledger location and lifecycle#

The ledger file’s location is the source’s choice – importers pass a path (conventionally beside the data being imported). Note carefully: deleting output files does not reset the ledger. The ledger records what was ingested, not what exists downstream, so re-running an import after deleting its .bdf.csv output produces nothing until the ledger is cleared – with reset(), or battfeed import --reset-ledger on the CLI. A corrupt or wrong-shaped ledger file raises rather than being silently treated as empty (an empty ledger would re-ingest everything – duplicate data is data loss’s quieter sibling); delete the file to start fresh (--reset-ledger cannot repair it: the source typically opens the ledger before the reset hook can run).

Every mutation is persisted immediately with an atomic write-unique-temp-then-replace, so a crash can never leave a half-written (invalid JSON) ledger behind.

class battfeed.ingest_state.ImportLedger(path)[source]#

JSON-file-backed dedupe ledger and quarantine for batch import sources.

Typical use – hash each candidate file exactly once per poll, and defer record() into commit_batch() (the commit point; see the module docstring):

class FolderImporter:
    def __init__(self, directory):
        self.directory = Path(directory)
        self.ledger = ImportLedger(self.directory / ".import-ledger.json")
        self._pending = None  # (path, content_hash) awaiting commit

    def poll(self):
        for path in sorted(self.directory.glob("*.txt")):
            digest = self.ledger.hash_of(path)
            if digest is None:
                continue  # zero-byte: never content-keyed; re-judged next poll
            if self.ledger.seen(path, content_hash=digest):
                continue
            if self.ledger.is_quarantined(path, content_hash=digest):
                continue
            try:
                rows = parse(path)
            except UnsupportedFormat as exc:
                self.ledger.quarantine(path, str(exc), content_hash=digest)
                continue
            self._pending = (path, digest)
            return rows
        return []

    def commit_batch(self):
        # Called by run_import AFTER sink.write succeeded.
        if self._pending is not None:
            path, digest = self._pending
            self.ledger.record(path, content_hash=digest)
            self._pending = None

All queries and mutations are keyed by the file’s content hash (see the module docstring for why content, not path – and why zero-byte files are exempt). Pass content_hash= (from hash_of()) to avoid re-reading the file for every call; without it, each call hashes the file itself. The ledger file is created on the first mutation; constructing against a missing file is an empty ledger.

Concurrency: one writer at a time; concurrent imports sharing a ledger file lose updates (see the module docstring). This class is crash-safe, not multi-writer-safe.

Parameters:

path (str | Path) – The ledger file. Its parent directory is created on demand.

Raises:

ValueError – if an existing ledger file is not valid JSON, is valid JSON of the wrong shape, or has an unrecognised version – delete the file to start fresh (see the module docstring for why corruption is not silently ignored).

property path: Path#

The ledger file this instance persists to.

property ingested_count: int#

Number of distinct file contents recorded as ingested.

property quarantined_count: int#

Number of distinct file contents quarantined.

hash_of(path)[source]#

The sha256 content hash keying path, or None for a zero-byte file.

Hash once per file and pass the result to seen() / record() / quarantine() / is_quarantined() via content_hash= – each of those otherwise re-reads the whole file. None means the file is empty and never content-keyed (see the module docstring): the ledger will not remember anything about it.

Parameters:

path (str | Path)

Return type:

str | None

seen(path, *, content_hash=None)[source]#

Whether a file with this exact content has been recorded as ingested.

Logs at info whenever it answers True, so a skipped file is visible in the run’s log rather than silently absent (invariant I2). Always False for zero-byte files.

Parameters:
Return type:

bool

record(path, *, content_hash=None)[source]#

Record the file’s content as ingested and persist. Idempotent.

Call this from commit_batch(), after the rows reached the sink – never from inside poll() (see the module docstring’s commit-point section). Declines (at info level) for zero-byte files.

Parameters:
Return type:

None

quarantine(path, reason, *, content_hash=None)[source]#

Mark the file’s content as permanently unsupported and persist.

reason is a human-readable explanation surfaced by quarantine_reason() – a quarantined file must explain itself (invariant I2). Re-quarantining the same content with the same reason is a quiet no-op; a changed reason rewrites the entry and warns again. Declines (at info level) for zero-byte files.

Parameters:
Return type:

None

is_quarantined(path, *, content_hash=None)[source]#

Whether a file with this exact content is quarantined.

Always False for zero-byte files (they are judged afresh each run). Logs at info when it answers True, so the skip is visible.

Parameters:
Return type:

bool

quarantine_reason(path, *, content_hash=None)[source]#

The recorded reason for a quarantined file, or None if not quarantined.

Parameters:
Return type:

str | None

reset()[source]#

Clear the ingested set and the quarantine, and persist the empty state.

This is what battfeed import --reset-ledger reaches through a source’s reset_ledger() hook. Deleting output files never resets the ledger; this does. It cannot repair a corrupt ledger file – loading one raises before any reset hook can run; delete the file as the error message directs.

Return type:

None

Sinks#

Write collected samples as a BDF (Battery Data Format) CSV file.

BDF files use snake_case machine-readable headers of the form {quantity}_{unit}. Every conforming file carries the required trio test_time_second, voltage_volt and current_ampere.

Sign convention (per the Battery Data Format specification, and used throughout battfeed): positive current charges the test object (current flows into it); negative current discharges it. Power follows the same sign as current.

battfeed.sinks.bdf_csv.REQUIRED_COLUMNS: tuple[str, ...] = ('test_time_second', 'voltage_volt', 'current_ampere')#

The trio every BDF file must contain, in the order they lead the header.

class battfeed.sinks.bdf_csv.BdfCsvSink(path, *, columns=None, metadata=None, clock=<built-in function monotonic>)[source]#

Stream samples into a .bdf.csv file with a JSON metadata sidecar.

The header always leads with the required trio test_time_second, voltage_volt, current_ampere (in that order) followed by any extra columns sorted alphabetically. If columns is not given, the column set is inferred from the first batch written; later samples with unknown extra keys are dropped from the file (with a debug log), and samples missing a column leave that cell empty.

The routing keys in battfeed.RESERVED_KEYS (series_id / run_id) are stripped from every row and from any explicit column set before header inference and writing, so a routing-aware source wired directly to this plain sink can never leak them into CSV columns.

A sidecar <name>.meta.json (the .bdf.csv suffix replaced) is written next to the data file. It contains the metadata mapping plus the started/finished timestamps, the battfeed version, the column list, the row count, and a finalized flag. Credentials are masked before writing: any secret-named entry (see battfeed.config.is_secret_key()) at any depth becomes ***, and scheme://user:pass@host userinfo in any string value is masked – so a credential a source echoes in its metadata() never reaches disk. (The CLI additionally scrubs known secret values; the sink is the value-independent chokepoint.) To survive a crash mid-collection the sidecar is written early – as soon as the data file is first opened (finalized: false) – rewritten periodically as rows accumulate, and rewritten a final time on close() with finalized: true and the final row count. A sidecar with finalized: false therefore marks a data file whose collection did not finish cleanly.

Parameters:

  • path (str | Path) – Output file path, conventionally named via dataset_filename().

  • columns (Sequence[str] | None) – Optional explicit column set (order-insensitive; the header ordering rule above is applied regardless).

  • metadata (Mapping[str, Any] | None) – Optional JSON-serialisable mapping recorded in the sidecar (operator, cell id, instrument settings, …).

  • clock (Callable[[], float]) – Monotonic clock used only to pace mid-collection sidecar rewrites; injectable so tests can drive it without wall time.

write(rows)[source]#

Append a batch of samples to the file (opens it on first use).

Reserved routing keys (series_id / run_id) are stripped from every row before header inference and writing.

Parameters:

rows (Iterable[Mapping[str, float | int | str]])

Return type:

None

close()[source]#

Close the CSV file and finalise the .meta.json sidecar. Idempotent.

Return type:

None

battfeed.sinks.bdf_csv.dataset_filename(institution, cell_name, date, seq)[source]#

Build a BDF dataset file name: InstitutionCode__CellName__YYYYMMDD_XXX.bdf.csv.

Example:

>>> dataset_filename("SINTEF", "CR2032-01", datetime.date(2026, 7, 7), 3)
'SINTEF__CR2032-01__20260707_003.bdf.csv'
Parameters:
Return type:

str

battfeed.sinks.bdf_csv.validate_file(path)[source]#

Validate an emitted file with the batterydf package (optional extra).

Returns the validation report dict from bdf.validate (it contains at least an "ok" boolean). battfeed itself never parses or normalises vendor data; this simply hands the finished file to the reference implementation of the format.

Raises:

  • ImportError – if batterydf is not installed – install it with pip install "battfeed[bdf]".

  • RuntimeError – if batterydf fails while validating the file.

Parameters:

path (str | Path)

Return type:

dict[str, Any]

Demultiplex one sample stream into one BDF file per (series, run), with rotation.

BDF’s invariant is one test object, one monotonic timebase, per file – but the world violates it in three directions: one connection can yield many objects (an account with N cars, a drone with N packs), one object can yield many runs (a pack flies many flights, each restarting its clock), and many sources never end at all (a shunt streams forever). RoutingSink resolves all three at the sink layer:

  • Samples carrying the reserved routing keys (series_id / run_id, see battfeed.RESERVED_KEYS) are demultiplexed into one child BdfCsvSink per (series_id, run_id). Samples without series_id flow to a single default stream, so routing-free sources work unchanged.

  • A new run_id for a known series closes the previous file and opens the next one – runs never merge into a non-monotonic timebase.

  • Rotation (rotate_after_s / rotate_after_rows, whichever trips first) does the same without a run_id change: segments are runs for endless streams, turning unbounded telemetry into a sequence of bounded, valid BDF files (each finalized with its sidecar as soon as it rotates, so a crash loses at most the open segments – and even those keep the early, unfinalised sidecar the child sink writes at open time).

Filenames follow the BDF convention via dataset_filename(); the _XXX sequence slot advances per run/segment for the same series on the same day. Raw series_id values are device serials, not filenames – they may contain __ (reserved as the BDF filename separator), path separators, characters Windows forbids, or be empty – so they pass through sanitize_cell_name(), with deterministic hash-suffix disambiguation when two distinct ids sanitize to the same name.

Timebase ownership (invariant I5). This sink routes and never restamps: test_time_second is written exactly as the source supplied it. A source that emits routing keys must therefore supply its own test_time_second, zero-based per (series, run) – the harvester’s shared elapsed-collection stamp is wrong for an object that appears mid-run (see battfeed.protocols).

class battfeed.sinks.routing.RoutingSink(directory, *, institution='LOCAL', series_info=None, metadata=None, rotate_after_s=None, rotate_after_rows=None, sink_factory=<class 'battfeed.sinks.bdf_csv.BdfCsvSink'>, clock=<built-in function monotonic>, today=<built-in method today of type object>)[source]#

Route one sample stream into one BDF file per (series, run), rotating.

Implements the battfeed.Sink protocol, so it drops in anywhere a plain battfeed.BdfCsvSink does – the harvester never knows the difference. Child files open lazily as series/runs appear, so a series that comes online mid-stream gets its own file and sidecar from its first sample; closing a segment (run change, rotation, or close()) finalizes that segment’s sidecar immediately.

Per-segment sidecar metadata is the shared metadata mapping, overlaid with the series_info metadata for the series, overlaid with {"series_id": ..., "run_id": ..., "segment": n} (segment counts from 1 per series).

Parameters:

  • directory (str | Path) – Directory the .bdf.csv files (and their sidecars) are written into; created on demand.

  • institution (str) – Institution code for dataset_filename().

  • series_info (Callable[[str], tuple[str, Mapping[str, Any]]] | None) – Optional callable(series_id) -> (cell_name, metadata_dict) hook so lazily-discovered objects get proper filenames and their own sidecar content. Called once per new series; without it the sanitized series_id names the cell.

  • metadata (Mapping[str, Any] | None) – Shared base metadata recorded in every segment’s sidecar.

  • rotate_after_s (float | None) – Close and re-open a series’ file once it has been open this many seconds (measured with clock).

  • rotate_after_rows (int | None) – Close and re-open a series’ file once it holds this many rows. With both limits set, whichever trips first rotates.

  • sink_factory (Callable[..., Sink]) – callable(path, *, metadata) -> Sink building each child sink; injectable so tests can capture routed rows in memory. (Path allocation still reserves each claimed path on disk as an empty file – see _next_path() – regardless of the factory.)

  • clock (Callable[[], float]) – Monotonic clock driving time-based rotation; injectable for tests (see the Harvester’s clock/sleep injection).

  • today (Callable[[], datetime.date]) – Date provider for filenames; injectable so tests get deterministic names.

Unlike BdfCsvSink, closing without ever writing produces no files: there is nothing to route, so nothing is (even emptily) recorded.

property files_by_series: dict[str | None, list[Path]]#

Paths written so far, keyed by series_id (None = default stream).

Segments appear in the order they were opened; the last entry of a list may still be open. Returns a copy.

write(rows)[source]#

Route a batch of samples to their per-(series, run) child sinks.

Reserved routing keys are stripped from every row before delegation; everything else – including any source-supplied test_time_second – is passed through untouched (invariant I5: the source owns the per-(series, run) timebase).

Parameters:

rows (Iterable[Mapping[str, float | int | str]])

Return type:

None

close()[source]#

Close every open child sink (finalising its sidecar). Idempotent.

Return type:

None

battfeed.sinks.routing.sanitize_cell_name(raw)[source]#

Turn a raw series id (e.g. a device serial) into a safe BDF cell name.

Raw ids are whatever the device reports: they may contain __ (reserved as the BDF filename separator), path separators, characters illegal on Windows, control characters, or be empty. The result is deterministic, non-empty, at most 60 characters, and never contains __ – nor starts or ends with _ (which would recreate __ next to the filename separators).

Distinct raw ids can sanitize to the same name ("pack/1" and "pack?1" both become "pack-1", and names differing only by case collide too – Windows filesystems are case-insensitive); RoutingSink disambiguates such collisions with a stable hash suffix derived from the raw id.

Parameters:

raw (str)

Return type:

str

Push collected samples to any HTTP endpoint that accepts NDJSON.

A generic ingest client for community registries, lab servers, and custom platforms: rows are buffered, then POSTed as newline-delimited JSON (optionally gzipped) on a time-based cadence. Stdlib only.

Delivery semantics#

The sink guarantees at-least-once delivery of every accepted row: a failed POST keeps the whole buffer for the next flush, rows still unsent when HttpPushSink.close() gives up are spooled to disk as a loadable .spool.ndjson file, and a buffer that outgrows max_buffered_rows spills its oldest rows to a spool file instead of dropping them. Servers deduplicate however they choose (row content, timestamps, an id column of their own); the sink makes no exactly-once claim.

Serialisation#

Every row is serialised to one RFC 8259-valid JSON object per line: non-finite floats (NaN, +/-inf – sensor dropout) become null and are counted and warned about, bytes become base64 strings, datetime / date values become ISO 8601 strings, and any other non-JSON type falls back to str(). The literal NaN / Infinity tokens the stdlib would otherwise emit are rejected by strict parsers, so they never reach the wire or a spool file.

Unlike battfeed.BdfCsvSink – which strips the reserved routing keys because they are never BDF columns – this sink includes series_id / run_id in the payload: a receiving server needs them to demultiplex one stream into per-object, per-run storage.

class battfeed.sinks.http_push.HttpPushSink(url, *, token=None, headers=None, batch_seconds=15.0, timeout=10.0, compress=True, spool_dir=None, close_retry_delays=(2.0, 4.0), max_buffered_rows=100000, clock=<built-in function monotonic>, sleep=<built-in function sleep>)[source]#

Buffer samples and POST them as newline-delimited JSON to url.

Each row becomes one JSON object per line, with keys exactly as the sample provides them. The reserved routing keys (series_id / run_id) are deliberately included – in deliberate contrast to battfeed.BdfCsvSink, which must strip them – because they are meaningful to a receiving server for demultiplexing one stream into per-object, per-run storage.

Network trouble never propagates out of write(): the harvester’s ErrorPolicy governs sources, not sinks – a sink heals itself. A failed or non-2xx POST keeps the whole buffer for the next flush (logged at warning with counts), so delivery is at-least-once; servers deduplicate however they choose. close() retries the final flush and spools any remaining rows to disk rather than dropping them.

Parameters:

  • url (str) – Endpoint accepting POST bodies of newline-delimited JSON. Userinfo and query strings are redacted from every log message.

  • token (str | None) – Optional bearer token, sent as Authorization: Bearer <token>. Rejected at construction if it contains \r or \n.

  • headers (Mapping[str, str] | None) – Optional extra headers, merged last (the caller wins over every generated header, including Authorization). Names and values are rejected at construction if they contain \r/\n.

  • batch_seconds (float) – Cadence of time-based flushing on the injected clock; write() triggers a flush once this much time has passed since the last attempt.

  • timeout (float) – Per-request socket timeout in seconds.

  • compress (bool) – Gzip the request body (Content-Encoding: gzip).

  • spool_dir (str | Path | None) – Directory for the .spool.ndjson file written when close() cannot deliver the remaining rows (or when the buffer overflows max_buffered_rows). Defaults to the current working directory so at-least-once holds unconfigured; if the directory is unusable, the current working directory is the fallback.

  • close_retry_delays (Sequence[float]) – Seconds slept between close-time flush attempts; close() makes len(close_retry_delays) + 1 attempts.

  • max_buffered_rows (int) – Upper bound on buffered rows. When exceeded, the oldest rows are spilled to a spool file immediately (warned, and counted in records_spooled) so memory stays bounded while at-least-once is preserved.

  • clock (Callable[[], float]) – Monotonic clock used to pace flushing; injectable for tests.

  • sleep (Callable[[float], None]) – Sleep function used between close-time retries; injectable.

records_sent#

Rows acknowledged with a 2xx response so far.

records_spooled#

Rows written to spool files (close-time spooling plus buffer-overflow spills).

write(rows)[source]#

Buffer a batch of samples; flush when batch_seconds has passed.

Never raises on network trouble – a failed flush keeps the buffer and the next cadence tick tries again. A buffer that outgrows max_buffered_rows spills its oldest rows to a spool file.

Parameters:

rows (Iterable[Mapping[str, float | int | str]])

Return type:

None

flush()[source]#

POST the buffered rows as (optionally gzipped) NDJSON.

Returns True when the buffer is empty afterwards (nothing to send, or the server answered 2xx and the buffer was cleared). Any other response, a network error, or a request-building error keeps the whole buffer for the next flush and returns False; nothing is raised.

Return type:

bool

close()[source]#

Final flush with retries, then spool whatever remains. Idempotent.

Makes len(close_retry_delays) + 1 flush attempts (sleeping the configured delays between them via the injected sleep); rows still undelivered are written to <spool_dir>/<host>-<utcstamp>.spool.ndjson – one JSON object per line, loadable for later re-send – and counted in records_spooled, never dropped. With the default delays and timeout this blocks at most ~36 s against a hung server (three attempts x 10 s timeout, plus 2 s + 4 s of sleep) and ~6 s against one that refuses connections promptly.

Return type:

None

Write collected samples as a single Parquet file (optional extra).

Parquet is an analytics format, not BDF: the reserved routing keys (series_id / run_id) are kept as ordinary columns so downstream dataframe work can group by object and run – in deliberate contrast to battfeed.BdfCsvSink, which must strip them.

Deliberately simple: rows are buffered in memory and one file is written on ParquetSink.close() – no row-group tuning, no append mode, no rotation. Long-running rotation belongs to RoutingSink + BDF files; Parquet is for bounded analytical captures.

Requires the optional pyarrow dependency:

pip install "battfeed[parquet]"
class battfeed.sinks.parquet.ParquetSink(path, *, metadata=None)[source]#

Buffer samples in memory and write one Parquet file on close.

The column set is the union of the keys of every row written, in first-seen order; rows missing a column get a null in that cell. The reserved routing keys (series_id / run_id) are included as ordinary columns – Parquet is an analytics format, not BDF.

A <name>.meta.json sidecar is written next to the data file in the same format battfeed.BdfCsvSink uses (metadata, started/finished timestamps, battfeed version, columns, row count, finalized: true). Because everything is written at close, there is no early/unfinalised sidecar stage: this sink is for bounded analytical captures, not unbounded telemetry – use RoutingSink + BDF for rotation.

Parameters:

  • path (str | Path) – Output .parquet file path.

  • metadata (Mapping[str, Any] | None) – Optional JSON-serialisable mapping recorded in the sidecar.

Raises:

ImportError – on construction, if pyarrow is not installed.

write(rows)[source]#

Buffer a batch of samples (nothing touches disk until close).

Parameters:

rows (Iterable[Mapping[str, float | int | str]])

Return type:

None

close()[source]#

Write the Parquet file and its .meta.json sidecar. Idempotent.

Never raises and never loses the capture: if the table build or the file write fails (e.g. ArrowInvalid from mixed-type columns), every buffered row is rescued to <stem>.rescue.ndjson beside the target (one RFC-valid JSON object per line, non-finite floats as null) and the sidecar is written with finalized: false plus an error field naming the failure. The data file itself is written via a temporary file and os.replace, so a failed write leaves no partial file at the final path.

Return type:

None

battfeed.sources.streaming: push-style hardware base#

StreamingSource: adapt push-style delivery to the synchronous poll() seam.

Many integrations do not answer questions – they talk: BLE notification callbacks, CAN frames, MQTT messages. The battfeed contract, on the other hand, is deliberately pull-based (battfeed.DataSource.poll()), because one synchronous seam keeps every source testable and every pipeline identical. StreamingSource bridges the two: a subclass supplies a blocking run_reader() loop, the base runs it in a daemon thread, and readings accumulate in a bounded buffer that poll() drains.

Three design points carry the invariants and are worth spelling out:

  • The buffer is bounded, and overflow is counted (invariant I2 – no silent data loss). An unbounded buffer would grow without limit whenever the device outpaces the poll loop – a months-long feed would fail by memory exhaustion at the worst possible moment. So the buffer is a deque(maxlen=buffer_size), which keeps the newest readings. A plain deque drops the oldest entry silently, though, which would be exactly the silent loss I2 forbids – so every overflow increments dropped_total, is surfaced through stream_stats() / metadata(), and emits a rate-limited warning.

  • Reader errors surface at ``poll()``, and a dead device keeps FAILING (invariant I4 – sources raise, the harvester owns retry). A background thread that dies quietly would leave a source that “collects” nothing forever. An exception that escapes run_reader is parked and re-raised by the next poll(); the poll after that starts a replacement reader session. The trap in that design is subtle: if the restarting poll simply returned [] it would count as a success, reset the harvester’s consecutive-failure counter, and a permanently dead device would alternate failure with fabricated success forever – SourceFailure unreachable, backoff pinned at its minimum. So restart polls are honest instead: while the previous session ended without delivering a single sample, the restarting poll starts the replacement and raises DeadReaderError noting how many sessions in a row died empty. During a dead-device period every poll therefore raises, the harvester’s counter genuinely accumulates with escalating backoff, and the default battfeed.ErrorPolicy reaches battfeed.SourceFailure. A session that delivers at least one sample resets the escalation, so a flaky-but-working device is retried indefinitely, exactly as intended. No retry loop belongs inside run_reader.

  • Sessions have generations, so zombies cannot contaminate (I2 again, from the other side – no *fabricated* data either). close() joins the reader with a timeout; a reader stuck in a blocking wait that ignores should_stop is abandoned rather than blocking the caller forever. Every session is stamped with a generation number, and emits or death exceptions arriving from a stale generation are discarded (with a debug log) – an abandoned zombie waking up later can neither inject samples into a newer session’s stream nor fail it with a stale error.

The DataSource protocol itself is unchanged – third parties still need no battfeed import. Subclassing this base is merely the convenient way to satisfy the protocol for push-style hardware; replay fixtures for developing subclasses without hardware live in battfeed.testing.replay.

exception battfeed.sources.streaming.DeadReaderError(source, dead_sessions, last_error)[source]#

The reader keeps ending without delivering a single sample.

Raised by StreamingSource.poll() while it starts a replacement reader session, whenever every session since the last delivered sample has died empty. This keeps a dead device failing at the harvester on every poll – a restart that returned [] would be counted as a success and reset the battfeed.ErrorPolicy consecutive-failure counter, making battfeed.SourceFailure unreachable. A fresh instance is raised per restart; last_error carries the most recent exception a session died with (None when sessions returned cleanly but empty).

Parameters:
Return type:

None

class battfeed.sources.streaming.StreamingSource(name, *, buffer_size=4096, join_timeout_s=2.0, clock=<built-in function monotonic>)[source]#

Base class for push-style sources; subclasses implement run_reader().

The reader thread is started lazily on the first poll() and runs run_reader() until should_stop() turns True (set by close()) or the reader raises. Emitted samples land in a bounded buffer that poll() drains in arrival order. See the module docstring for why the buffer is bounded, why reader errors re-raise at poll(), and why restarts after sample-less sessions raise DeadReaderError instead of fabricating an empty success.

Lifecycle: close() is idempotent and restartable – a later poll() starts a fresh reader session with a clean escalation history (mirroring Mc3000Source, where a poll after close() reconnects). A reader that returns on its own (e.g. a scan that ended) is likewise restarted by the next poll – silently when the session delivered samples, via DeadReaderError when it ended empty. Every start after the first is counted in reader_restarts.

Thread-safety: poll() and close() may be called from any thread (concurrent callers are serialized internally), and emit may be called from any reader-side callback.

Parameters:

  • name (str) – Source name (the DataSource.name seam attribute).

  • buffer_size (int) – Maximum samples buffered between polls. When the reader outpaces the poll loop, the oldest buffered samples are dropped and counted in dropped_total.

  • join_timeout_s (float) – How long close() waits for the reader thread to honor should_stop before abandoning it (the daemon thread cannot keep the process alive, and an abandoned session’s late contributions are discarded by the generation guard).

  • clock (Callable[[], float]) – Monotonic clock, injectable for tests (used only to rate-limit overflow warnings – samples are never timestamped here; timebase policy belongs to the subclass or the harvester).

run_reader(emit, should_stop)[source]#

Blocking receive loop; subclasses must override.

Call emit(sample) for every received reading (thread-safe; may be called from callbacks) and return promptly once should_stop() is True – check it between blocking waits, or use it to cancel them. Raise on trouble instead of retrying: the exception is re-raised by the next poll() and the harvester’s error policy owns the retry (invariant I4). Raise a fresh exception per crash – the base defensively clears a parked exception’s __traceback__, but a cached instance re-raised from several places accumulates state and confuses whoever reads the eventual stack trace.

Parameters:
Return type:

None

metadata()[source]#

Base metadata contribution: identity plus the streaming stats.

Subclasses normally extend it: {**super().metadata(), ...}. The stats ride along so every sidecar records whether the buffer ever overflowed (invariant I2).

Return type:

Mapping[str, Any]

poll()[source]#

Drain and return the buffered samples, in arrival order.

If the reader session has died with an exception since the last poll, that exception is re-raised here (invariant I4) and no restart happens yet. The next poll starts a replacement session – silently when the dead session had delivered at least one sample, otherwise that poll also raises DeadReaderError, so a dead device fails on every poll and the harvester’s error policy genuinely escalates (see the module docstring). The very first start never raises. Buffered samples survive raising polls and are returned once polling resumes.

Return type:

list[dict[str, float | int | str]]

close()[source]#

Stop the reader session. Idempotent; a later poll() reconnects.

Waits join_timeout_s for the reader to honor should_stop, then abandons it: the generation guard discards anything a lingering zombie later emits or dies with. Closing also resets the dead-session escalation, so the post-close reconnect starts with a clean history (like the very first start).

Return type:

None

property received_total: int#

Samples accepted from live reader sessions (including any later dropped).

property dropped_total: int#

Samples lost to buffer overflow so far (oldest-first; invariant I2).

property reader_restarts: int#

Reader session starts beyond the first (crash recoveries and reconnects).

property reader_alive: bool#

True while the current session’s thread is running.

A session abandoned by close() (join timeout) may physically linger, but it no longer counts here and the generation guard keeps it from contributing anything.

Type:

Best-effort

stream_stats()[source]#

One consistent snapshot of the counters (also merged into metadata).

Return type:

dict[str, int]

battfeed.testing: the source author’s toolkit#

Contract test kit: assert that a source honors the DataSource seam.

battfeed’s universality depends on third parties shipping sources, and the seam is structural (battfeed.DataSource is a typing.Protocol) – nothing forces an implementation to be correct at import time. This module is the executable half of the contract: call check_source() from your own test suite against your source (backed by a mock transport or a replay tape, never live hardware) and it asserts the essentials every battfeed pipeline relies on.

What is checked#

The structural surface (name / metadata() / poll() and the optional close() / availability() hooks); strict JSON serializability of metadata and samples (allow_nan=False – NaN and Infinity are rejected, because sidecars and BDF consumers cannot represent them); sample shape (non-empty str keys; int/float/str values; bool is rejected everywhere – isinstance(True, int) holds in Python, but a bool in a measurement column is always a bug); test_time_second numeric and non-negative; freshness (the same dict object must not appear twice in one batch or be recycled across polls – downstream code mutates rows in place); and the reserved-key discipline: a sample carrying series_id / run_id must also carry its own test_time_second (invariant I5 – the harvester’s shared elapsed-collection stamp is wrong for objects that appear mid-run).

Deliberately NOT checked#

Honesty about the kit’s blind spots, so a green check is not oversold:

  • Blocking polls – poll() must not block for longer than roughly one polling interval; a checker cannot draw that line for hardware it has never seen.

  • Duplicate-content batches – rows that are equal by value are legal (an instrument may genuinely repeat a reading); only object identity (aliasing) is checked.

  • Mutation of previously returned rows – a source that hands out fresh dicts but later mutates the old ones passes; catching it would require deep-copy snapshots of every batch, and downstream consumers should not be reading old batches anyway.

  • Cross-poll ``test_time_second`` monotonicity – runs may legitimately restart the clock at segment boundaries; validating monotonicity requires run semantics the kit cannot know. Validate emitted files with batterydf instead.

  • Shared-timebase I5-in-spirit violations – a routing source that stamps every series from one shared clock satisfies the per-sample rule checked here while still violating I5’s intent; only a test with two series appearing at different times can catch that.

  • Column vocabulary, sign conventions, units – these need domain knowledge of your device; verify signs against real charge/discharge behavior and validate files with batterydf.

Failures raise AssertionError with a message naming the violated rule, so a bare check_source(MySource(...)) inside any test function is a complete contract test.

battfeed.testing.contract.check_source(source, *, polls=3)[source]#

Assert the DataSource protocol essentials on a live instance.

Polls the source polls times, so drive it with a mock transport or a replay tape – never hardware. If the source has a close() hook it is called (twice – battfeed’s own sources document close() as idempotent, and the CLI relies on that being safe). See the module docstring for the full list of what is checked and, just as important, what is deliberately not checked (blocking polls, duplicate-content batches, cross-poll test_time_second monotonicity, shared-timebase I5 violations, vocabulary/signs/units).

Parameters:

  • source (Any) – The instance to check (any object; failures explain what is missing).

  • polls (int) – Number of poll() calls to sample-check (>= 1).

Raises:

AssertionError – with a message naming the violated contract rule.

Return type:

None

Record/replay tapes: hardware-free fixtures for streaming sources.

Streaming hardware (BLE advertisements, CAN frames, MQTT messages) is awkward to develop against – the primary development machine is Windows while the tooling is Linux-first, devices are not on every desk, and CI has neither. The replay-first workflow: record raw frames from a live session once into a tape, commit the tape, and drive every test and most development from the tape with no hardware and no waiting.

Tape format#

A tape is a JSONL file (one JSON object per line, UTF-8), stable and diff-friendly on purpose – tapes are committed fixtures:

{"t": 0.0,   "data": "10099f...", "meta": {"rssi": -61}}
{"t": 1.024, "data": "10099e..."}

  • t – seconds since the start of the recording (float, >= 0, non-decreasing).

  • data – the raw frame bytes, hex-encoded. What a “frame” is belongs to the recording source (a BLE advertisement payload, a CAN frame, an MQTT message body); the tape does not interpret it.

  • meta – optional JSON object of per-frame context (RSSI, CAN arbitration id, topic, …). Anonymize identifiers (MAC addresses, keys) before committing a field recording.

Replaying compresses time by default: ReplayReader delivers frames as fast as the consumer accepts them, so an hour-long tape replays in milliseconds – the offsets stay available on each frame for sources that derive test_time_second from them. Pacing (for demos or soak tests) is opt-in and clock-injectable, so even paced tests can run on a fake clock with zero real sleeps.

class battfeed.testing.replay.ReplayReader(tape, *, pace=None, clock=<built-in function monotonic>, sleep=<built-in function sleep>)[source]#

Deliver a tape’s frames to a handler – the reader loop of a replay source.

Built to slot straight into battfeed.StreamingSource.run_reader(): pass the base’s should_stop through and decode each frame into emit:

class ReplayedShuntSource(StreamingSource):
    def __init__(self, tape: ReplayTape) -> None:
        super().__init__("replayed-shunt")
        self._tape = tape

    def run_reader(self, emit, should_stop):
        ReplayReader(self._tape).run(
            lambda frame: emit(self._decode(frame)), should_stop
        )

By default (pace=None) frames are delivered as fast as the handler accepts them – time compression: an hour of recording replays in milliseconds with zero sleeps, and the recorded offset stays available as frame.t for sources that derive test_time_second from it. pace=1.0 replays on the recorded timeline (2.0 = twice as fast, …) using the injectable clock/sleep, so even paced replay is testable on a fake clock without real waiting.

Parameters:
run(handle, should_stop=None)[source]#

Feed every frame to handle; return the number delivered.

Checks should_stop before each frame (and before each paced sleep), so a StreamingSource shutdown ends the replay promptly.

Parameters:
Return type:

int

class battfeed.testing.replay.ReplayTape(frames=())[source]#

An ordered sequence of TapeFrame with JSONL load/save.

Frame offsets must be >= 0 and non-decreasing however the tape is built (constructor, append() or load()) – a replayed timeline that jumps backwards would silently corrupt any test_time_second derived from it.

Parameters:

frames (Iterable[TapeFrame])

append(t, data, meta=None)[source]#

Append one frame; offsets must be >= 0 and non-decreasing.

meta is validated eagerly: it must be JSON-serializable with finite numbers (allow_nan=False), so a bad recording fails at the recording site instead of poisoning every later load. Note that JSON objects have string keys: an int key like {1: "a"} serializes fine but round-trips as {"1": "a"}.

Parameters:
Return type:

None

property duration_s: float#

Offset of the last frame (0.0 for an empty tape).

classmethod load(path)[source]#

Read a JSONL tape file; malformed lines raise with the line number.

The whole tape is read into memory – tapes are committed test fixtures, not archives; keep them small.

One concession to reality: recorders flush line by line, so an OS crash or power loss can leave a torn, half-written final line. A malformed last line is therefore skipped with a warning instead of voiding the tape; malformed lines anywhere else (and out-of-order offsets anywhere, including the last line) stay hard errors, because they mean corruption or hand-editing, not a torn tail.

Parameters:

path (str | Path)

Return type:

ReplayTape

save(path)[source]#

Write the tape as JSONL (the format committed fixtures use).

Parameters:

path (str | Path)

Return type:

None

class battfeed.testing.replay.TapeFrame(t, data, meta=None)[source]#

One recorded frame: a time offset, raw bytes, optional context.

Parameters:
t: float#

Seconds since the start of the recording.

data: bytes#

Raw frame bytes, exactly as received from the transport.

meta: Mapping[str, Any] | None = None#

Optional per-frame context (RSSI, arbitration id, topic, …).

class battfeed.testing.replay.TapeRecorder(path, *, clock=<built-in function monotonic>)[source]#

Tee raw frames from a live reader into a tape file, line by line.

Turning a field session into a committed fixture should cost one flag in the recording tool: wrap the existing frame callback with tee() (or call record() directly from it) and every frame is appended to the JSONL tape with its offset from the recorder’s start. Each line is flushed as written, so an interrupted session still leaves a loadable tape of everything received so far.

Usable as a context manager; close() is idempotent.

Parameters:

  • path (str | Path)

  • clock (Callable[[], float])

record(data, meta=None)[source]#

Append one frame, stamped with the offset since the recorder opened.

Parameters:
Return type:

None

tee(callback)[source]#

Wrap a live frame callback so every frame is recorded, then forwarded.

Parameters:

callback (Callable[[bytes], None])

Return type:

Callable[[bytes], None]

close()[source]#

Flush and close the tape file. Idempotent.

Return type:

None

On this page
Show Source