herosdevices.core.bus.onewire

This module provides a class for managing single device and bus-wide OneWire connections.

Module Contents

class herosdevices.core.bus.onewire.OneWire(device_id: str, sysfs_path: str = '/sys/bus/w1/')

A read-only Onewire driver that relies on the Linux kernel w1 driver.

Linux exposes onewire devices in sysfs. This driver can read the sysfs files as they are specified in the :param sensors: list.

sysfs_path

class herosdevices.core.bus.onewire.BusWideAction

Describes a bus-master-level operation that must be triggered once and drained.

Some onewire device families expose an operation on the bus master itself (rather than on the individual device) that must be triggered once and then “drained” by reading every affected device before the next trigger is meaningful. w1_therm’s bulk conversion (therm_bulk_read) converts all bound w1_therm devices at once, but the master’s status only returns to a settled state once every affected device has been read.

families: frozenset[str]

trigger_attr: str

trigger_value: str = 'trigger'

status_attr: str = ''

pending_value: str = '-1'

read_attr: str = 'temperature'

herosdevices.core.bus.onewire.THERM_BULK_READ

class herosdevices.core.bus.onewire.OneWireBusMaster(master_id: str = 'w1_bus_master1', sysfs_path: str = '/sys/bus/w1/', bulk_actions: collections.abc.Sequence[BusWideAction] = (THERM_BULK_READ,), status_timeout: float = 2.0)

Owns a single onewire bus and performs bus-wide actions on behalf of several consumers.

A single process should own one instance of this class and call poll_once() (or run_forever()) on an interval. Any number of consumer processes/threads can then read the most recently cached raw value for a device via read() without touching the bus themselves. This avoids two problems that arise from reading a onewire bus concurrently from many independent readers: needless bus contention from redundant per-device conversions, and, for families with a bulk-read mechanism, an unresolvable ambiguity about whether a given bulk-read status reflects the current trigger or a stale one.

The cache holds the raw sysfs string for each device, keyed by its own device_id (e.g. “28-00000df6f834”) - unconverted, with no caller-chosen naming layer. Unit conversion, staleness handling, and validity checks are left to consumers. Only devices whose family is covered by a bulk_actions entry are read and cached; the read_attr to read on the device comes from that entry, so no per-family device class or factory is needed. Use device_ids to discover which device_ids are relevant.

sysfs_path = '/sys/bus/w1/'

devices_root

master_path

bulk_actions

status_timeout = 2.0

property is_polling: bool

Return whether a poll cycle triggered by poll_once() is currently running.

property device_ids: collections.abc.Sequence[str]

Return every discovered device_id whose family is covered by a bulk action.

poll_once() → None

Trigger one poll cycle on a background thread; returns immediately.

Raises:

RuntimeError – if a previously triggered poll cycle is still running.

run_forever(interval: float) → None

Call poll_once() every interval seconds, isolating errors between cycles.

Since poll_once() is a non-blocking trigger, each iteration returns almost immediately; if a drain cycle takes longer than interval, poll_once() raises RuntimeError on the next iteration (already covered by this loop’s broad except) until the previous cycle finishes.

read(device_id: str) → str | None

Return the most recently cached raw value for device_id.

Returns None both when device_id was never read and when its last read failed - the cache does not distinguish the two, by design.

class herosdevices.core.bus.onewire.W1ThermBusMaster(master_id: str, sysfs_path: str = '/sys/bus/w1/', status_timeout: float = 2.0)

Bases: OneWireBusMaster

A OneWireBusMaster fixed to w1_therm’s bulk-read action (DS18S20/DS1822/DS18B20/DS1825/DS28EA00).

Use this instead of OneWireBusMaster directly when a bus is dedicated to w1_therm devices - it takes no bulk_actions argument, so it can be instantiated straight from a JSON device config (BusWideAction has no JSON representation). If a physical bus ever mixes w1_therm devices with another bulk-action family, construct OneWireBusMaster directly instead, passing both actions.

To support a new bulk-action family: define a BusWideAction constant next to THERM_BULK_READ, then add a subclass fixing bulk_actions to it, following this class as the template.

sysfs_path = '/sys/bus/w1/'

devices_root

master_path

bulk_actions

status_timeout = 2.0

property is_polling: bool

Return whether a poll cycle triggered by poll_once() is currently running.

property device_ids: collections.abc.Sequence[str]

Return every discovered device_id whose family is covered by a bulk action.

poll_once() → None

Trigger one poll cycle on a background thread; returns immediately.

Raises:

RuntimeError – if a previously triggered poll cycle is still running.

run_forever(interval: float) → None

Call poll_once() every interval seconds, isolating errors between cycles.

Since poll_once() is a non-blocking trigger, each iteration returns almost immediately; if a drain cycle takes longer than interval, poll_once() raises RuntimeError on the next iteration (already covered by this loop’s broad except) until the previous cycle finishes.

read(device_id: str) → str | None

Return the most recently cached raw value for device_id.

Returns None both when device_id was never read and when its last read failed - the cache does not distinguish the two, by design.

class herosdevices.core.bus.onewire.OneWireBusConsumer(bus_master: OneWireBusMaster, device_id: str)

A read-only onewire device that sources its raw value from a OneWireBusMaster instead of sysfs.

Behaves like OneWire._observable_data: converts the family’s _default_observables from a raw string into (value, unit) pairs. Where OneWire reads sysfs directly, this reads the most recently cached value for device_id from a shared, already-polled bus master - see OneWireBusMaster for why a single poller per physical bus is preferred over each consumer reading the bus independently.

bus_master

device_id

class herosdevices.core.bus.onewire.W1ThermBusConsumer(bus_master: OneWireBusMaster, device_id: str)

Bases: OneWireBusConsumer

w1_therm sensor, read from a shared OneWireBusMaster’s cache instead of sysfs directly.

bus_master

device_id