"""This module provides a class for managing single device and bus-wide OneWire connections."""
import re
import threading
import time
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from pathlib import Path
from herosdevices.helper import log
_FAMILY_SERIAL_RE = re.compile(r"^[0-9A-Fa-f]{2}-[0-9A-Fa-f]+$")
[docs]
class OneWire:
"""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.
"""
# _default_observables describes which quantities can be extracted from the onewire device. The list entries
# have the form (name_of_observable, conversion_function, unit). The "name_of_observable" must be an endpoint in
# the sysfs directory of the onewire device.
_default_observables: list[tuple[str, Callable, str]] = []
def __init__(self, device_id: str, sysfs_path: str = "/sys/bus/w1/") -> None:
"""
Initialize a onewire connection to a device.
Args:
device_id: id of the onewire device as named by the w1 Linux kernel driver.
sysfs_path: sys path of the Linux w1 kernel driver
"""
self.sysfs_path = Path(sysfs_path) / "devices" / device_id
if not Path(self.sysfs_path).exists():
log.error(f"Can not access Linux w1 sysfs path {sysfs_path}")
def _observable_data(self) -> dict[str, tuple[float, str]]:
try:
return {
observable: (conversion(self._read_content(self.sysfs_path / observable)), unit)
for observable, conversion, unit in self._default_observables
}
except ValueError:
return {}
def _read_content(self, path: Path) -> str | None:
"""Read Contents of :param path: (str) filename."""
try:
with path.open("r") as file:
return "".join(file.readlines())
except (OSError, TypeError, ValueError):
log.warning(f"Could not read file {path}")
return None
[docs]
@dataclass(frozen=True)
class 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"
def __post_init__(self) -> None:
"""Default status_attr to trigger_attr when not given explicitly."""
if not self.status_attr:
object.__setattr__(self, "status_attr", self.trigger_attr)
# DS18S20, DS1822, DS18B20, DS1825, DS28EA00 - all bound to the w1_therm driver and drained via
# the bus master's therm_bulk_read entry.
THERM_BULK_READ = BusWideAction(
families=frozenset({"10", "22", "28", "3B", "42"}),
trigger_attr="therm_bulk_read",
read_attr="temperature",
)
[docs]
class OneWireBusMaster:
"""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 :meth:`poll_once` (or
:meth:`run_forever`) on an interval. Any number of consumer processes/threads can then read
the most recently cached raw value for a device via :meth:`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 :attr:`device_ids` to
discover which device_ids are relevant.
"""
def __init__(
self,
master_id: str = "w1_bus_master1",
sysfs_path: str = "/sys/bus/w1/",
bulk_actions: Sequence[BusWideAction] = (THERM_BULK_READ,),
status_timeout: float = 2.0,
) -> None:
"""
Initialize a onewire bus master.
Args:
master_id: id of the onewire bus master as named by the w1 Linux kernel driver
(e.g. "w1_bus_master1").
sysfs_path: sys path of the Linux w1 kernel driver.
bulk_actions: bus-wide actions to trigger and drain once per poll, keyed by the device
families they cover. Devices whose family isn't covered by any bulk_actions entry
are not read or cached.
status_timeout: seconds to wait for a bulk action's status to leave pending_value
before giving up on that action for the current poll cycle.
"""
self.sysfs_path = sysfs_path
self.devices_root = Path(sysfs_path) / "devices"
self.master_path = self.devices_root / master_id
if not self.master_path.exists():
log.error(f"Can not access Linux w1 sysfs path {self.master_path}")
self.bulk_actions = list(bulk_actions)
self.status_timeout = status_timeout
self._cache: dict[str, str | None] = {}
self._poll_lock = threading.Lock()
self._poll_thread: threading.Thread | None = None
self._last_poll_duration: float | None = None
@property
def is_polling(self) -> bool:
"""Return whether a poll cycle triggered by poll_once() is currently running."""
return self._poll_thread is not None and self._poll_thread.is_alive()
@staticmethod
def _family(device_id: str) -> str:
"""Extract the family code (prefix before the dash) from a device_id."""
return device_id.split("-", 1)[0]
def _is_covered(self, device_id: str) -> bool:
"""Return whether device_id's family is covered by any bulk_actions entry."""
family = self._family(device_id)
return any(family in action.families for action in self.bulk_actions)
def _discover_devices(self) -> dict[str, Path]:
"""Return every onewire device_id currently present on the bus, mapped to its sysfs path."""
try:
return {entry.name: entry for entry in self.devices_root.iterdir() if _FAMILY_SERIAL_RE.match(entry.name)}
except OSError:
log.warning(f"Could not list devices under {self.devices_root}")
return {}
@property
def device_ids(self) -> Sequence[str]:
"""Return every discovered device_id whose family is covered by a bulk action."""
return tuple(device_id for device_id in self._discover_devices() if self._is_covered(device_id))
@staticmethod
def _read_raw(path: Path) -> str | None:
"""Read the raw contents of a sysfs attribute file, or None on failure."""
try:
return path.read_text().strip()
except OSError:
return None
def _drain_action(self, action: BusWideAction, device_ids: Sequence[str]) -> None:
"""Trigger a bus-wide action and read every affected device to leave the master settled."""
trigger_path = self.master_path / action.trigger_attr
status_path = self.master_path / action.status_attr
try:
# some sysfs store handlers silently ignore a write missing the trailing newline instead of
# matching it as the expected command, so always send one regardless of trigger_value.
trigger_path.write_text(f"{action.trigger_value}\n")
except OSError:
log.warning(f"Could not trigger bus-wide action via {trigger_path}")
# Devices covered by this action may still hold a stale conversion result; read them
# anyway so a failed trigger doesn't leave the master permanently unsettled.
else:
deadline = time.monotonic() + self.status_timeout
while time.monotonic() < deadline:
if self._read_raw(status_path) != action.pending_value:
break
time.sleep(0.05)
else:
log.warning(f"Timed out waiting for {status_path} to leave pending state")
for device_id in device_ids:
self._read_device(device_id, action.read_attr)
def _read_device(self, device_id: str, read_attr: str) -> None:
"""Read a single device's raw attribute and cache it under its device_id."""
path = self.devices_root / device_id / read_attr
raw = self._read_raw(path)
if raw is None:
log.warning(f"Could not read file {path}")
self._cache[device_id] = raw
[docs]
def poll_once(self) -> None:
"""Trigger one poll cycle on a background thread; returns immediately.
Raises:
RuntimeError: if a previously triggered poll cycle is still running.
"""
with self._poll_lock:
if self.is_polling:
msg = "OneWireBusMaster poll already in progress"
raise RuntimeError(msg)
self._poll_thread = threading.Thread(target=self._poll, daemon=True)
self._poll_thread.start()
def _poll(self) -> None:
"""Run one full poll cycle: trigger and drain every bulk action's covered devices."""
start = time.monotonic()
try:
discovered = self._discover_devices()
for action in self.bulk_actions:
relevant = [device_id for device_id in discovered if self._family(device_id) in action.families]
if not relevant:
continue
self._drain_action(action, relevant)
except Exception: # noqa: BLE001
log.exception("Unhandled error during onewire poll cycle")
finally:
self._last_poll_duration = time.monotonic() - start
[docs]
def run_forever(self, interval: float) -> None:
"""Call :meth:`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.
"""
while True:
start = time.monotonic()
try:
self.poll_once()
except Exception: # noqa: BLE001
log.exception("Unhandled error during onewire poll cycle")
elapsed = time.monotonic() - start
time.sleep(max(0.0, interval - elapsed))
[docs]
def read(self, 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.
"""
return self._cache.get(device_id)
def _observable_data(self) -> dict[str, tuple[float | None, str]]:
"""Trigger a poll cycle and report the duration of the most recently completed one.
Non-blocking: triggers the next poll_once() cycle on a background thread and immediately
returns the previous cycle's duration (None before the first cycle has completed).
Propagates poll_once()'s RuntimeError if a previous poll is still running - callers such as
heros's PolledLocalDatasourceHERO already log and ignore such errors.
"""
self.poll_once()
return {"poll_time": (self._last_poll_duration, "s")}
[docs]
class W1ThermBusMaster(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.
"""
def __init__(
self,
master_id: str,
sysfs_path: str = "/sys/bus/w1/",
status_timeout: float = 2.0,
) -> None:
"""
Initialize a onewire bus master fixed to the w1_therm bulk-read action.
Args:
master_id: id of the onewire bus master as named by the w1 Linux kernel driver
(e.g. "w1_bus_master1").
sysfs_path: sys path of the Linux w1 kernel driver.
status_timeout: seconds to wait for the bulk-read status to leave pending_value before
giving up for the current poll cycle.
"""
super().__init__(
master_id=master_id,
sysfs_path=sysfs_path,
bulk_actions=(THERM_BULK_READ,),
status_timeout=status_timeout,
)
[docs]
class OneWireBusConsumer:
"""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.
"""
_default_observables: list[tuple[str, Callable, str]] = []
def __init__(self, bus_master: OneWireBusMaster, device_id: str) -> None:
"""
Initialize a onewire device backed by a bus master's cache.
Args:
bus_master: the OneWireBusMaster (typically a RemoteHERO proxy to one) that polls the
physical bus this device lives on.
device_id: id of the onewire device as named by the w1 Linux kernel driver
(e.g. "28-00000df6f834").
"""
self.bus_master = bus_master
self.device_id = device_id
def _observable_data(self) -> dict[str, tuple[float, str]]:
try:
return {
observable: (conversion(self.bus_master.read(self.device_id)), unit)
for observable, conversion, unit in self._default_observables
}
except ValueError:
return {}
[docs]
class W1ThermBusConsumer(OneWireBusConsumer):
"""w1_therm sensor, read from a shared OneWireBusMaster's cache instead of sysfs directly."""
_default_observables = [("temperature", float, "mdegC")]