`watlowlib.registry`¶

The cross-protocol parameter registry — PARAMETERS, ParameterRegistry, ParameterSpec, RwesFlag, family table, enumerations, and unit enums. See Parameters and Design §5a.

Public surface¶

watlowlib.registry ¶

Parameter and family registry.

The registry is the cross-protocol seam: each parameter row carries both Std Bus selector (cls / member / instance) and Modbus selector (relative_addr / absolute_addr / register_count) metadata so command variants can lower a single read_parameter(...) call to either protocol with no per-parameter bespoke code. See docs/design.md §5a.

ControllerFamily ¶

Bases: StrEnum

Watlow controller family discriminator.

Membership here is advisory — :class:watlowlib.devices.session.Session treats family hints as priors, not gates. See docs/design.md §5b.

ParameterRegistry ¶

ParameterRegistry(
    specs,
    *,
    aliases=DEFAULT_ALIASES,
    name_overrides=_NAME_OVERRIDES,
)

Indexed view over a sequence of :class:ParameterSpec rows.

Lookups are O(1) on canonical name, alias, and parameter_id. Construction is O(N).

Source code in src/watlowlib/registry/parameters.py

def __init__(
    self,
    specs: tuple[ParameterSpec, ...],
    *,
    aliases: Mapping[str, str] = DEFAULT_ALIASES,
    name_overrides: Mapping[int, str] = _NAME_OVERRIDES,
) -> None:
    # ``name_overrides`` promotes parameter ids to short canonical
    # names. The PM registry uses the bundled :data:`_NAME_OVERRIDES`
    # (its auto-derived names are verbose); the SD registry passes
    # ``{}`` and instead carries verbatim ``canonical`` names baked
    # into each spec at load time. Passing the PM table to a non-PM
    # registry would mis-claim collision keys (e.g. "units") for PM
    # parameter ids that don't exist in that registry.
    # Apply name overrides + collect aliases per spec.
    rebound: list[ParameterSpec] = []
    for spec in specs:
        override = name_overrides.get(spec.parameter_id)
        name = override or spec.name
        spec_aliases: set[str] = set()
        for alias, target in aliases.items():
            if target == name:
                spec_aliases.add(alias)
        if override and override != spec.name:
            # Keep the original auto-generated name as an alias so
            # callers that learn it from the JSON still resolve.
            spec_aliases.add(spec.name)
        if name != spec.name or spec_aliases:
            # ``dataclasses.replace`` copies every other field
            # verbatim — including any field added to ParameterSpec
            # later (``scale``, future per-row overrides). A manual
            # field-by-field rebind would silently drop new fields,
            # so never reintroduce one here.
            spec = replace(  # noqa: PLW2901 — frozen dataclass rebind
                spec,
                name=name,
                aliases=frozenset(spec_aliases),
            )
        rebound.append(spec)

    self._specs: tuple[ParameterSpec, ...] = tuple(rebound)
    # A handful of canonical names auto-derived by ``_canonical_name``
    # collide across unrelated rows (e.g. "Display - Units",
    # "Analog Input - Units", and "Linearization - Units" all
    # canonicalise to ``"units"``). When an :data:`_NAME_OVERRIDES`
    # entry targets one of those collision names, the override row
    # wins — otherwise the JSON's iteration order silently decided
    # which spec a public name like ``"units"`` resolved to.
    override_owner: dict[str, int] = {name.lower(): pid for pid, name in name_overrides.items()}
    by_id: dict[int, ParameterSpec] = {}
    by_name: dict[str, ParameterSpec] = {}
    for spec in self._specs:
        by_id[spec.parameter_id] = spec
        key = spec.name.lower()
        owner = override_owner.get(key)
        if owner is not None and owner != spec.parameter_id:
            # Another row's auto-canonical name collides with an
            # override-assigned name; the override row owns the key.
            continue
        by_name[key] = spec
        for alias in spec.aliases:
            by_name.setdefault(alias.lower(), spec)
    self._by_id: Mapping[int, ParameterSpec] = MappingProxyType(by_id)
    self._by_name: Mapping[str, ParameterSpec] = MappingProxyType(by_name)

has ¶

has(name_or_id)

Return True if name_or_id resolves; never raises.

Source code in src/watlowlib/registry/parameters.py

def has(self, name_or_id: str | int) -> bool:
    """Return ``True`` if ``name_or_id`` resolves; never raises."""
    try:
        self.resolve(name_or_id)
    except WatlowValidationError:
        return False
    return True

resolve ¶

resolve(name_or_id)

Look up a spec by canonical name, alias, or parameter ID.

Raises:

Type	Description
`WatlowValidationError`	`name_or_id` does not resolve.

Source code in src/watlowlib/registry/parameters.py

def resolve(self, name_or_id: str | int) -> ParameterSpec:
    """Look up a spec by canonical name, alias, or parameter ID.

    Raises:
        WatlowValidationError: ``name_or_id`` does not resolve.
    """
    if isinstance(name_or_id, int):
        try:
            return self._by_id[name_or_id]
        except KeyError as exc:
            raise WatlowValidationError(
                f"unknown parameter id: {name_or_id}",
            ) from exc
    key = name_or_id.lower()
    try:
        return self._by_name[key]
    except KeyError as exc:
        raise WatlowValidationError(
            f"unknown parameter name: {name_or_id!r}",
        ) from exc

validate_instance ¶

validate_instance(spec, instance)

Raise if instance is out of range for spec.

Public so the variant layer can validate before encoding.

Source code in src/watlowlib/registry/parameters.py

def validate_instance(self, spec: ParameterSpec, instance: int) -> None:
    """Raise if ``instance`` is out of range for ``spec``.

    Public so the variant layer can validate before encoding.
    """
    if instance < 1 or instance > spec.max_instance:
        raise WatlowValidationError(
            f"instance {instance} out of range for {spec.name!r} (1..{spec.max_instance})",
        )

validate_value ¶

validate_value(spec, value)

Soft range check based on the spec's parsed range metadata.

Skipped silently if range_min / range_max couldn't be parsed from the JSON range field — Watlow's range strings are not always machine-readable. STRING parameters are not range-checked.

Source code in src/watlowlib/registry/parameters.py

def validate_value(self, spec: ParameterSpec, value: float | int | str) -> None:
    """Soft range check based on the spec's parsed ``range`` metadata.

    Skipped silently if ``range_min`` / ``range_max`` couldn't be
    parsed from the JSON ``range`` field — Watlow's range strings
    are not always machine-readable. STRING parameters are not
    range-checked.
    """
    if isinstance(value, str):
        return
    if spec.range_min is None or spec.range_max is None:
        return
    v = float(value)
    if v < spec.range_min or v > spec.range_max:
        raise WatlowValidationError(
            f"value {value!r} out of range for {spec.name!r} "
            f"({spec.range_min}..{spec.range_max})",
        )

ParameterSpec `dataclass` ¶

ParameterSpec(
    parameter_id,
    name,
    aliases,
    data_type,
    unit_kind,
    rwes,
    safety,
    cls,
    member,
    default_instance,
    max_instance,
    relative_addr,
    absolute_addr,
    register_count,
    word_order=None,
    range_min=None,
    range_max=None,
    default=None,
    family_hints=_empty_family_hints(),
    scale=1.0,
)

A single parameter row from pm_parameters.json.

Per-protocol fields:

Std Bus selector: :attr:cls, :attr:member, :attr:default_instance, :attr:max_instance.
Modbus selector: :attr:relative_addr, :attr:absolute_addr, :attr:register_count, :attr:word_order (None → client default HIGH_LOW per design §5a).

scale `class-attribute` `instance-attribute` ¶

scale = 1.0

Engineering-unit scale factor for the Modbus decode / encode path.

The wire stores raw integers (e.g. the Series SD reports a process value of 68421 for 68.421 °F); scale is the multiplier that turns the raw word into engineering units on read (value * scale) and the divisor that turns engineering units back into raw words on write (round(value / scale)).

1.0 (the default) means no scaling — and is applied as a strict identity: the read path skips the multiply entirely when scale == 1.0 so an integer parameter stays an int rather than being promoted to float by int * 1.0. Std Bus rows are never scaled (the Std Bus variant ignores this field).

RwesFlag ¶

Bases: StrEnum

Persistence + access flag from the EZ-ZONE register list.

R — read-only.
W — write-only (rare; typically actions like "start autotune").
RW — runtime read/write, not EEPROM-backed.
RWE — RW + persisted to EEPROM.
RWES — RWE + saved set ("save settings to user memory").

Mapping to :class:SafetyTier is in :func:_safety_from_rwes; the registry binds the result to :attr:ParameterSpec.safety at load time.

Unit ¶

Bases: StrEnum

Concrete display unit attached to a :class:Reading value.

UnitKind ¶

Bases: StrEnum

Structural unit family of a parameter, as declared by the registry.

Maps to a concrete :class:Unit at read time via :func:resolve_unit. TEMPERATURE resolves to °C or °F depending on the device's comms display setting (parameter 17050); the rest are independent of device state.

classify_family ¶

classify_family(part_number)

Return the :class:ControllerFamily for a part-number string.

Only the leading family discriminator is parsed; per-family digit decoding is in :func:decode_part_number.

Source code in src/watlowlib/registry/families.py

def classify_family(part_number: str) -> ControllerFamily:
    """Return the :class:`ControllerFamily` for a part-number string.

    Only the leading family discriminator is parsed; per-family digit
    decoding is in :func:`decode_part_number`.
    """
    head = part_number.strip().upper()
    if head.startswith("PM"):
        return ControllerFamily.PM
    if head.startswith("RM"):
        return ControllerFamily.RM
    if head.startswith("ST"):
        return ControllerFamily.ST
    if head.startswith("F4T"):
        return ControllerFamily.F4T
    if head.startswith("SD"):
        return ControllerFamily.SD
    return ControllerFamily.UNKNOWN

coerce_unit ¶

coerce_unit(value: Unit) -> Unit

coerce_unit(value: str) -> Unit

coerce_unit(value)

Normalise a :class:Unit-or-string into a :class:Unit.

Case-insensitive on the string side. Raises :class:WatlowValidationError on an unknown alias so the setter can fail pre-I/O before any wire bytes go out.

Raw integer device codes (15, 30) are not accepted — callers who want the lower-level path use write_parameter("display_units", 30).

Source code in src/watlowlib/registry/units.py

def coerce_unit(value: object) -> Unit:
    """Normalise a :class:`Unit`-or-string into a :class:`Unit`.

    Case-insensitive on the string side. Raises
    :class:`WatlowValidationError` on an unknown alias so the setter
    can fail pre-I/O before any wire bytes go out.

    Raw integer device codes (15, 30) are **not** accepted — callers
    who want the lower-level path use
    ``write_parameter("display_units", 30)``.
    """
    if isinstance(value, Unit):
        return value
    if not isinstance(value, str):
        raise WatlowValidationError(
            f"unit must be a Unit or string alias, got {type(value).__name__}",
        )
    alias = value.strip().lower()
    try:
        return _UNIT_STRING_ALIASES[alias]
    except KeyError as exc:
        raise WatlowValidationError(
            f"unknown unit alias: {value!r}",
        ) from exc

display_code_for_unit ¶

display_code_for_unit(unit)

Return the raw 17050 device code for a temperature display unit.

Source code in src/watlowlib/registry/units.py

def display_code_for_unit(unit: Unit) -> int | None:
    """Return the raw 17050 device code for a temperature display unit."""
    return _DISPLAY_CODE_FOR_UNIT.get(unit)

load_enumerations ¶

load_enumerations()

Load and return all symbol rows from enumerations.json.

Section-header rows (where the value column is a string) are dropped; only (_, _, _, int) rows survive.

Source code in src/watlowlib/registry/enumerations.py

def load_enumerations() -> tuple[EnumerationRow, ...]:
    """Load and return all symbol rows from ``enumerations.json``.

    Section-header rows (where the value column is a string) are
    dropped; only ``(_, _, _, int)`` rows survive.
    """
    raw = files(_DATA_PACKAGE).joinpath(_FILENAME).read_text(encoding="utf-8")
    blob: list[dict[str, list[Any]]] = json.loads(raw)
    out: list[EnumerationRow] = []
    for entry in blob:
        row = entry.get("row")
        if row is None or len(row) != _ROW_COLUMN_COUNT:
            continue
        value = row[3]
        if not isinstance(value, int):
            # Header rows carry a string in the value column.
            continue
        out.append((row[0], str(row[1]), row[2], value))
    return tuple(out)

load_parameters ¶

load_parameters(filename, *, family, family_hints=None)

Load every parameter spec from a bundled registry JSON file.

Parameters:

Name	Type	Description	Default
`filename`	`str`	Bare filename inside the :mod:`watlowlib.data` package (e.g. `"pm_parameters.json"` / `"sd_parameters.json"`).	required
`family`	`ControllerFamily`	The controller family this file describes. Used as the default single-member `family_hints` when `family_hints` is not given.	required
`family_hints`	`frozenset[ControllerFamily] \| None`	Explicit family-hint set stamped on every produced spec. Defaults to `frozenset({family})`.	`None`

Returns:

Name	Type	Description
`One`	`ParameterSpec`	class:`ParameterSpec` per loadable row, first occurrence
	`...`	winning on duplicate `parameter_id` (PM Map 1 / Map 2 sheets
	`tuple[ParameterSpec, ...]`	repeat ids with differing instance metadata).

Source code in src/watlowlib/registry/parameters.py

def load_parameters(
    filename: str,
    *,
    family: ControllerFamily,
    family_hints: frozenset[ControllerFamily] | None = None,
) -> tuple[ParameterSpec, ...]:
    """Load every parameter spec from a bundled registry JSON file.

    Args:
        filename: Bare filename inside the :mod:`watlowlib.data` package
            (e.g. ``"pm_parameters.json"`` / ``"sd_parameters.json"``).
        family: The controller family this file describes. Used as the
            default single-member ``family_hints`` when ``family_hints``
            is not given.
        family_hints: Explicit family-hint set stamped on every produced
            spec. Defaults to ``frozenset({family})``.

    Returns:
        One :class:`ParameterSpec` per loadable row, first occurrence
        winning on duplicate ``parameter_id`` (PM Map 1 / Map 2 sheets
        repeat ids with differing instance metadata).
    """
    hints = family_hints if family_hints is not None else frozenset({family})
    raw_text = files(_DATA_PACKAGE).joinpath(filename).read_text(encoding="utf-8")
    rows: list[dict[str, Any]] = json.loads(raw_text)
    out: list[ParameterSpec] = []
    seen_ids: set[int] = set()
    for raw in rows:
        spec = _build_spec(raw, family_hints=hints)
        if spec is None:
            continue
        if spec.parameter_id in seen_ids:
            # Duplicates across PM Map 1 / Map 2 sheets are real (same
            # parameter_id, different instance metadata) — keep only
            # the first occurrence; the registry exposes max_instance
            # so callers reach all loops.
            continue
        seen_ids.add(spec.parameter_id)
        out.append(spec)
    return tuple(out)

load_pm_parameters ¶

load_pm_parameters()

Load and return every PM parameter spec from the bundled JSON.

Source code in src/watlowlib/registry/parameters.py

def load_pm_parameters() -> tuple[ParameterSpec, ...]:
    """Load and return every PM parameter spec from the bundled JSON."""
    return load_parameters(_PM_FILENAME, family=ControllerFamily.PM)

load_sd_parameters ¶

load_sd_parameters()

Load and return every Series SD parameter spec from the bundled JSON.

Source code in src/watlowlib/registry/parameters.py

def load_sd_parameters() -> tuple[ParameterSpec, ...]:
    """Load and return every Series SD parameter spec from the bundled JSON."""
    return load_parameters(_SD_FILENAME, family=ControllerFamily.SD)

resolve_unit ¶

resolve_unit(kind, temperature_unit)

Resolve a parameter's :class:UnitKind to a concrete :class:Unit.

TEMPERATURE → temperature_unit (passes the caller's asserted wire scale through, or None when none was asserted).
PERCENT → :attr:Unit.PERCENT.
Everything else → None.

Pure mapping; no I/O. The caller (typically :class:watlowlib.devices.session.Session) is responsible for determining the wire scale and passing it in.

Source code in src/watlowlib/registry/units.py

def resolve_unit(kind: UnitKind, temperature_unit: Unit | None) -> Unit | None:
    """Resolve a parameter's :class:`UnitKind` to a concrete :class:`Unit`.

    - ``TEMPERATURE`` → ``temperature_unit`` (passes the caller's
      asserted wire scale through, or ``None`` when none was asserted).
    - ``PERCENT`` → :attr:`Unit.PERCENT`.
    - Everything else → ``None``.

    Pure mapping; no I/O. The caller (typically
    :class:`watlowlib.devices.session.Session`) is responsible for
    determining the wire scale and passing it in.
    """
    if kind is UnitKind.TEMPERATURE:
        return temperature_unit
    if kind is UnitKind.PERCENT:
        return Unit.PERCENT
    return None

unit_from_display_code ¶

unit_from_display_code(code)

Return the display unit for a raw 17050 device code, if known.

Source code in src/watlowlib/registry/units.py

def unit_from_display_code(code: int) -> Unit | None:
    """Return the display unit for a raw 17050 device code, if known."""
    return _DISPLAY_UNIT_CODES.get(code)

Parameter specs + registry¶

watlowlib.registry.parameters ¶

Parameter registry — the cross-protocol seam.

Each row of data/pm_parameters.json is loaded once into a :class:ParameterSpec, indexed by canonical name (with aliases) and by parameter_id. The spec carries enough information to lower a read_parameter("setpoint") call to either Std Bus or Modbus with no per-parameter bespoke code.

Loading is eager: a module-level :data:PARAMETERS is built at import time so subsequent lookups are O(1) dict reads. Loading is also fail-loud for malformed rows — a row missing decode metadata for its declared :class:DataType (e.g. a PACKED row with no count) is not silently dropped; it is surfaced as an :class:watlowlib.errors.WatlowProtocolError at load time.

ParameterRegistry ¶

ParameterRegistry(
    specs,
    *,
    aliases=DEFAULT_ALIASES,
    name_overrides=_NAME_OVERRIDES,
)

Indexed view over a sequence of :class:ParameterSpec rows.

Lookups are O(1) on canonical name, alias, and parameter_id. Construction is O(N).

Source code in src/watlowlib/registry/parameters.py

def __init__(
    self,
    specs: tuple[ParameterSpec, ...],
    *,
    aliases: Mapping[str, str] = DEFAULT_ALIASES,
    name_overrides: Mapping[int, str] = _NAME_OVERRIDES,
) -> None:
    # ``name_overrides`` promotes parameter ids to short canonical
    # names. The PM registry uses the bundled :data:`_NAME_OVERRIDES`
    # (its auto-derived names are verbose); the SD registry passes
    # ``{}`` and instead carries verbatim ``canonical`` names baked
    # into each spec at load time. Passing the PM table to a non-PM
    # registry would mis-claim collision keys (e.g. "units") for PM
    # parameter ids that don't exist in that registry.
    # Apply name overrides + collect aliases per spec.
    rebound: list[ParameterSpec] = []
    for spec in specs:
        override = name_overrides.get(spec.parameter_id)
        name = override or spec.name
        spec_aliases: set[str] = set()
        for alias, target in aliases.items():
            if target == name:
                spec_aliases.add(alias)
        if override and override != spec.name:
            # Keep the original auto-generated name as an alias so
            # callers that learn it from the JSON still resolve.
            spec_aliases.add(spec.name)
        if name != spec.name or spec_aliases:
            # ``dataclasses.replace`` copies every other field
            # verbatim — including any field added to ParameterSpec
            # later (``scale``, future per-row overrides). A manual
            # field-by-field rebind would silently drop new fields,
            # so never reintroduce one here.
            spec = replace(  # noqa: PLW2901 — frozen dataclass rebind
                spec,
                name=name,
                aliases=frozenset(spec_aliases),
            )
        rebound.append(spec)

    self._specs: tuple[ParameterSpec, ...] = tuple(rebound)
    # A handful of canonical names auto-derived by ``_canonical_name``
    # collide across unrelated rows (e.g. "Display - Units",
    # "Analog Input - Units", and "Linearization - Units" all
    # canonicalise to ``"units"``). When an :data:`_NAME_OVERRIDES`
    # entry targets one of those collision names, the override row
    # wins — otherwise the JSON's iteration order silently decided
    # which spec a public name like ``"units"`` resolved to.
    override_owner: dict[str, int] = {name.lower(): pid for pid, name in name_overrides.items()}
    by_id: dict[int, ParameterSpec] = {}
    by_name: dict[str, ParameterSpec] = {}
    for spec in self._specs:
        by_id[spec.parameter_id] = spec
        key = spec.name.lower()
        owner = override_owner.get(key)
        if owner is not None and owner != spec.parameter_id:
            # Another row's auto-canonical name collides with an
            # override-assigned name; the override row owns the key.
            continue
        by_name[key] = spec
        for alias in spec.aliases:
            by_name.setdefault(alias.lower(), spec)
    self._by_id: Mapping[int, ParameterSpec] = MappingProxyType(by_id)
    self._by_name: Mapping[str, ParameterSpec] = MappingProxyType(by_name)

has ¶

has(name_or_id)

Return True if name_or_id resolves; never raises.

Source code in src/watlowlib/registry/parameters.py

def has(self, name_or_id: str | int) -> bool:
    """Return ``True`` if ``name_or_id`` resolves; never raises."""
    try:
        self.resolve(name_or_id)
    except WatlowValidationError:
        return False
    return True

resolve ¶

resolve(name_or_id)

Look up a spec by canonical name, alias, or parameter ID.

Raises:

Type	Description
`WatlowValidationError`	`name_or_id` does not resolve.

Source code in src/watlowlib/registry/parameters.py

def resolve(self, name_or_id: str | int) -> ParameterSpec:
    """Look up a spec by canonical name, alias, or parameter ID.

    Raises:
        WatlowValidationError: ``name_or_id`` does not resolve.
    """
    if isinstance(name_or_id, int):
        try:
            return self._by_id[name_or_id]
        except KeyError as exc:
            raise WatlowValidationError(
                f"unknown parameter id: {name_or_id}",
            ) from exc
    key = name_or_id.lower()
    try:
        return self._by_name[key]
    except KeyError as exc:
        raise WatlowValidationError(
            f"unknown parameter name: {name_or_id!r}",
        ) from exc

validate_instance ¶

validate_instance(spec, instance)

Raise if instance is out of range for spec.

Public so the variant layer can validate before encoding.

Source code in src/watlowlib/registry/parameters.py

def validate_instance(self, spec: ParameterSpec, instance: int) -> None:
    """Raise if ``instance`` is out of range for ``spec``.

    Public so the variant layer can validate before encoding.
    """
    if instance < 1 or instance > spec.max_instance:
        raise WatlowValidationError(
            f"instance {instance} out of range for {spec.name!r} (1..{spec.max_instance})",
        )

validate_value ¶

validate_value(spec, value)

Soft range check based on the spec's parsed range metadata.

Skipped silently if range_min / range_max couldn't be parsed from the JSON range field — Watlow's range strings are not always machine-readable. STRING parameters are not range-checked.

Source code in src/watlowlib/registry/parameters.py

def validate_value(self, spec: ParameterSpec, value: float | int | str) -> None:
    """Soft range check based on the spec's parsed ``range`` metadata.

    Skipped silently if ``range_min`` / ``range_max`` couldn't be
    parsed from the JSON ``range`` field — Watlow's range strings
    are not always machine-readable. STRING parameters are not
    range-checked.
    """
    if isinstance(value, str):
        return
    if spec.range_min is None or spec.range_max is None:
        return
    v = float(value)
    if v < spec.range_min or v > spec.range_max:
        raise WatlowValidationError(
            f"value {value!r} out of range for {spec.name!r} "
            f"({spec.range_min}..{spec.range_max})",
        )

ParameterSpec `dataclass` ¶

ParameterSpec(
    parameter_id,
    name,
    aliases,
    data_type,
    unit_kind,
    rwes,
    safety,
    cls,
    member,
    default_instance,
    max_instance,
    relative_addr,
    absolute_addr,
    register_count,
    word_order=None,
    range_min=None,
    range_max=None,
    default=None,
    family_hints=_empty_family_hints(),
    scale=1.0,
)

A single parameter row from pm_parameters.json.

Per-protocol fields:

Std Bus selector: :attr:cls, :attr:member, :attr:default_instance, :attr:max_instance.
Modbus selector: :attr:relative_addr, :attr:absolute_addr, :attr:register_count, :attr:word_order (None → client default HIGH_LOW per design §5a).

scale `class-attribute` `instance-attribute` ¶

scale = 1.0

Engineering-unit scale factor for the Modbus decode / encode path.

The wire stores raw integers (e.g. the Series SD reports a process value of 68421 for 68.421 °F); scale is the multiplier that turns the raw word into engineering units on read (value * scale) and the divisor that turns engineering units back into raw words on write (round(value / scale)).

1.0 (the default) means no scaling — and is applied as a strict identity: the read path skips the multiply entirely when scale == 1.0 so an integer parameter stays an int rather than being promoted to float by int * 1.0. Std Bus rows are never scaled (the Std Bus variant ignores this field).

RwesFlag ¶

Bases: StrEnum

Persistence + access flag from the EZ-ZONE register list.

R — read-only.
W — write-only (rare; typically actions like "start autotune").
RW — runtime read/write, not EEPROM-backed.
RWE — RW + persisted to EEPROM.
RWES — RWE + saved set ("save settings to user memory").

Mapping to :class:SafetyTier is in :func:_safety_from_rwes; the registry binds the result to :attr:ParameterSpec.safety at load time.

load_parameters ¶

load_parameters(filename, *, family, family_hints=None)

Load every parameter spec from a bundled registry JSON file.

Parameters:

Name	Type	Description	Default
`filename`	`str`	Bare filename inside the :mod:`watlowlib.data` package (e.g. `"pm_parameters.json"` / `"sd_parameters.json"`).	required
`family`	`ControllerFamily`	The controller family this file describes. Used as the default single-member `family_hints` when `family_hints` is not given.	required
`family_hints`	`frozenset[ControllerFamily] \| None`	Explicit family-hint set stamped on every produced spec. Defaults to `frozenset({family})`.	`None`

Returns:

Name	Type	Description
`One`	`ParameterSpec`	class:`ParameterSpec` per loadable row, first occurrence
	`...`	winning on duplicate `parameter_id` (PM Map 1 / Map 2 sheets
	`tuple[ParameterSpec, ...]`	repeat ids with differing instance metadata).

Source code in src/watlowlib/registry/parameters.py

def load_parameters(
    filename: str,
    *,
    family: ControllerFamily,
    family_hints: frozenset[ControllerFamily] | None = None,
) -> tuple[ParameterSpec, ...]:
    """Load every parameter spec from a bundled registry JSON file.

    Args:
        filename: Bare filename inside the :mod:`watlowlib.data` package
            (e.g. ``"pm_parameters.json"`` / ``"sd_parameters.json"``).
        family: The controller family this file describes. Used as the
            default single-member ``family_hints`` when ``family_hints``
            is not given.
        family_hints: Explicit family-hint set stamped on every produced
            spec. Defaults to ``frozenset({family})``.

    Returns:
        One :class:`ParameterSpec` per loadable row, first occurrence
        winning on duplicate ``parameter_id`` (PM Map 1 / Map 2 sheets
        repeat ids with differing instance metadata).
    """
    hints = family_hints if family_hints is not None else frozenset({family})
    raw_text = files(_DATA_PACKAGE).joinpath(filename).read_text(encoding="utf-8")
    rows: list[dict[str, Any]] = json.loads(raw_text)
    out: list[ParameterSpec] = []
    seen_ids: set[int] = set()
    for raw in rows:
        spec = _build_spec(raw, family_hints=hints)
        if spec is None:
            continue
        if spec.parameter_id in seen_ids:
            # Duplicates across PM Map 1 / Map 2 sheets are real (same
            # parameter_id, different instance metadata) — keep only
            # the first occurrence; the registry exposes max_instance
            # so callers reach all loops.
            continue
        seen_ids.add(spec.parameter_id)
        out.append(spec)
    return tuple(out)

load_pm_parameters ¶

load_pm_parameters()

Load and return every PM parameter spec from the bundled JSON.

Source code in src/watlowlib/registry/parameters.py

def load_pm_parameters() -> tuple[ParameterSpec, ...]:
    """Load and return every PM parameter spec from the bundled JSON."""
    return load_parameters(_PM_FILENAME, family=ControllerFamily.PM)

load_sd_parameters ¶

load_sd_parameters()

Load and return every Series SD parameter spec from the bundled JSON.

Source code in src/watlowlib/registry/parameters.py

def load_sd_parameters() -> tuple[ParameterSpec, ...]:
    """Load and return every Series SD parameter spec from the bundled JSON."""
    return load_parameters(_SD_FILENAME, family=ControllerFamily.SD)

Family classification + part-number decoder¶

watlowlib.registry.families ¶

Controller families and per-family part-number decoders.

The :class:ControllerFamily enum is the family discriminator used across the library. :func:classify_family parses the leading characters of a part number to that enum; :func:decode_part_number runs the family's full decoder when one is registered.

The EZ-ZONE PM decoder is the only full per-family decoder today. Other families fall through to the discriminator-only stub.

ControllerFamily ¶

Bases: StrEnum

Watlow controller family discriminator.

Membership here is advisory — :class:watlowlib.devices.session.Session treats family hints as priors, not gates. See docs/design.md §5b.

capabilities_for_part_number ¶

capabilities_for_part_number(part)

Decode :class:Capability bits from a parsed :class:PartNumber.

PM is the only family decoded today; other families return the family prior unchanged. Bits derived here are facts about the SKU — they do not depend on the device responding to any particular query, only on the part-number string.

Returns the family prior OR-ed with any decoded bits, so callers can use this as the authoritative seed for :attr:DeviceInfo.capabilities after :meth:Controller.identify captures the part number.

Source code in src/watlowlib/registry/families.py

def capabilities_for_part_number(part: PartNumber) -> Capability:
    """Decode :class:`Capability` bits from a parsed :class:`PartNumber`.

    PM is the only family decoded today; other families return the
    family prior unchanged. Bits derived here are *facts about the
    SKU* — they do not depend on the device responding to any
    particular query, only on the part-number string.

    Returns the family prior OR-ed with any decoded bits, so callers
    can use this as the authoritative seed for
    :attr:`DeviceInfo.capabilities` after :meth:`Controller.identify`
    captures the part number.
    """
    # Imported here to keep families.py a leaf module — the capability
    # enum lives under devices/ but families/ is depended on by
    # registry/parameters.py at import time.
    from watlowlib.devices.capability import (  # noqa: PLC0415
        Capability as _Capability,
    )
    from watlowlib.devices.capability import (  # noqa: PLC0415
        capabilities_for_family,
    )

    caps = capabilities_for_family(part.family)
    if part.family is not ControllerFamily.PM:
        return caps

    output_2 = part.details.get("output_2", "")
    if output_2 and output_2 != _PM_OUTPUT_NONE:
        caps |= _Capability.HAS_COOLING

    control = part.details.get("control_type", "")
    if control in _PM_CONTROL_PROFILES:
        caps |= _Capability.HAS_PROFILES | _Capability.PROFILE

    code = pm_comms_code(part)
    if code is not None:
        if code in _PM_COMMS_MODBUS:
            caps |= _Capability.HAS_MODBUS
        if code in _PM_COMMS_BLUETOOTH:
            caps |= _Capability.HAS_BLUETOOTH
        if code in _PM_COMMS_ETHERNET:
            caps |= _Capability.HAS_ETHERNET

    return caps

classify_family ¶

classify_family(part_number)

Return the :class:ControllerFamily for a part-number string.

Only the leading family discriminator is parsed; per-family digit decoding is in :func:decode_part_number.

Source code in src/watlowlib/registry/families.py

def classify_family(part_number: str) -> ControllerFamily:
    """Return the :class:`ControllerFamily` for a part-number string.

    Only the leading family discriminator is parsed; per-family digit
    decoding is in :func:`decode_part_number`.
    """
    head = part_number.strip().upper()
    if head.startswith("PM"):
        return ControllerFamily.PM
    if head.startswith("RM"):
        return ControllerFamily.RM
    if head.startswith("ST"):
        return ControllerFamily.ST
    if head.startswith("F4T"):
        return ControllerFamily.F4T
    if head.startswith("SD"):
        return ControllerFamily.SD
    return ControllerFamily.UNKNOWN

decode_part_number ¶

decode_part_number(raw)

Decode raw into a populated :class:PartNumber.

Dispatches to the per-family decoder based on :func:classify_family. Families without a decoder fall through to a bare :class:PartNumber carrying only the family discriminator.

Source code in src/watlowlib/registry/families.py

def decode_part_number(raw: str) -> PartNumber:
    """Decode ``raw`` into a populated :class:`PartNumber`.

    Dispatches to the per-family decoder based on
    :func:`classify_family`. Families without a decoder fall through
    to a bare :class:`PartNumber` carrying only the family
    discriminator.
    """
    # Imported here so this module stays a dependency-light leaf — the
    # ``models`` module pulls a handful of registry types and we don't
    # want a circular import on the family enum.
    from watlowlib.devices.models import PartNumber  # noqa: PLC0415

    family = classify_family(raw)
    if family is ControllerFamily.PM:
        _, details = _decode_pm(raw)
        return PartNumber(raw=raw, family=family, details=MappingProxyType(details))
    return PartNumber(raw=raw, family=family, details=MappingProxyType({}))

default_loops ¶

default_loops(part)

Return the default loop count for the controller behind part.

Used by :class:Controller.identify to seed :attr:DeviceInfo.loops and by :meth:Controller.loop to validate the n argument. Returns 1 whenever the family or digits are unknown — never raises.

Source code in src/watlowlib/registry/families.py

def default_loops(part: PartNumber) -> int:
    """Return the default loop count for the controller behind ``part``.

    Used by :class:`Controller.identify` to seed
    :attr:`DeviceInfo.loops` and by :meth:`Controller.loop` to
    validate the ``n`` argument. Returns ``1`` whenever the family or
    digits are unknown — never raises.
    """
    if part.family is ControllerFamily.PM:
        case = part.details.get("case_size", "")
        ctrl = part.details.get("control_type", "")
        return _PM_LOOPS.get((case, ctrl), 1)
    # RM is multi-loop in production but no decoder is wired up yet;
    # default to 1 so callers don't get surprise behaviour.
    return 1

pm_comms_code ¶

pm_comms_code(part)

Return the position-8 comms character of a PM part number, or None.

Position 8 is the first character of the 7-char options block — e.g. PM3R1CA-AAAAAAA has comms code A (Standard Bus only). Returns None for non-PM families and for PM part numbers without a parsed options string.

Source code in src/watlowlib/registry/families.py

def pm_comms_code(part: PartNumber) -> str | None:
    """Return the position-8 comms character of a PM part number, or ``None``.

    Position 8 is the first character of the 7-char options block —
    e.g. ``PM3R1CA-AAAAAAA`` has comms code ``A`` (Standard Bus only).
    Returns ``None`` for non-PM families and for PM part numbers
    without a parsed options string.
    """
    if part.family is not ControllerFamily.PM:
        return None
    options = part.details.get("options", "")
    if not options:
        return None
    return options[0]

pm_comms_supports_modbus ¶

pm_comms_supports_modbus(part)

Whether the part's comms position-8 character carries Modbus.

Source code in src/watlowlib/registry/families.py

def pm_comms_supports_modbus(part: PartNumber) -> bool:
    """Whether the part's comms position-8 character carries Modbus."""
    code = pm_comms_code(part)
    return code is not None and code in _PM_COMMS_MODBUS

Enumerations¶

watlowlib.registry.enumerations ¶

Loader for data/enumerations.json.

The enumerations file groups the symbolic names Watlow uses for parameter values (heat algorithms, sensor types, alarm states, ...). It is shaped as a flat list of rows, where each row is one of:

a 4-tuple [7Seg, PC label, text enumeration, value] — the actual symbol entry
a 4-tuple where the last element is a string column header (e.g. [..., "Value"]) — section header rows; skipped on load.

This module only loads the table. Binding specific symbol groups to :class:watlowlib.registry.parameters.ParameterSpec.enum happens elsewhere, once families and per-parameter enum metadata are wired through.

load_enumerations ¶

load_enumerations()

Load and return all symbol rows from enumerations.json.

Section-header rows (where the value column is a string) are dropped; only (_, _, _, int) rows survive.

Source code in src/watlowlib/registry/enumerations.py

def load_enumerations() -> tuple[EnumerationRow, ...]:
    """Load and return all symbol rows from ``enumerations.json``.

    Section-header rows (where the value column is a string) are
    dropped; only ``(_, _, _, int)`` rows survive.
    """
    raw = files(_DATA_PACKAGE).joinpath(_FILENAME).read_text(encoding="utf-8")
    blob: list[dict[str, list[Any]]] = json.loads(raw)
    out: list[EnumerationRow] = []
    for entry in blob:
        row = entry.get("row")
        if row is None or len(row) != _ROW_COLUMN_COUNT:
            continue
        value = row[3]
        if not isinstance(value, int):
            # Header rows carry a string in the value column.
            continue
        out.append((row[0], str(row[1]), row[2], value))
    return tuple(out)

Aliases¶

watlowlib.registry.aliases ¶

Friendly aliases for canonical parameter names.

Aliases like pv → process_value and sp → setpoint back the public read_parameter("pv") / read_parameter("setpoint") entry points. The :class:watlowlib.registry.parameters.ParameterRegistry consults this table when resolving a string that doesn't match a canonical name directly. Aliases are case-insensitive.

Adding new aliases is non-breaking — registry resolution is lookup, not generative.

Units¶

watlowlib.registry.units ¶

Unit vocabulary for Watlow parameters.

Two enums:

:class:Unit — the concrete unit a temperature/percent value is reported in (°C / °F / %). Attached to :class:Reading.unit and :class:Sample.unit.
:class:UnitKind — the structural unit family of a parameter as declared in the registry JSON (temperature / percent / dimensionless / enumeration / string). Used by :func:resolve_unit to compute the concrete :class:Unit for a temperature parameter given the (separately-determined) wire scale.

Watlow PM controllers expose two display-unit registers — 3005 ("Display - Units", front panel) and 17050 ("Communications - Display Units"). On at least one PM3 firmware (id 5678), 17050 is label- only: writing it changes the enum reported when 17050 is read back but does not change the scale of values exchanged over comms. The internal storage unit (the scale temperatures actually travel in over the wire) is governed by something else — and on devices where it cannot be determined empirically, the library refuses to guess.

Consequence for this module: :func:resolve_unit no longer assumes 17050 is the wire scale. The caller (the session) supplies an explicit temperature_unit derived from the assert_wire_temperature_unit user-assertion (or None when no assertion was made). Reading.unit = None is the honest answer for temperature reads when the wire scale is unknown.

See docs/devices.md §Units for the user-facing contract.

Unit ¶

Bases: StrEnum

Concrete display unit attached to a :class:Reading value.

UnitKind ¶

Bases: StrEnum

Structural unit family of a parameter, as declared by the registry.

Maps to a concrete :class:Unit at read time via :func:resolve_unit. TEMPERATURE resolves to °C or °F depending on the device's comms display setting (parameter 17050); the rest are independent of device state.

coerce_unit ¶

coerce_unit(value: Unit) -> Unit

coerce_unit(value: str) -> Unit

coerce_unit(value)

Normalise a :class:Unit-or-string into a :class:Unit.

Case-insensitive on the string side. Raises :class:WatlowValidationError on an unknown alias so the setter can fail pre-I/O before any wire bytes go out.

Raw integer device codes (15, 30) are not accepted — callers who want the lower-level path use write_parameter("display_units", 30).

Source code in src/watlowlib/registry/units.py

def coerce_unit(value: object) -> Unit:
    """Normalise a :class:`Unit`-or-string into a :class:`Unit`.

    Case-insensitive on the string side. Raises
    :class:`WatlowValidationError` on an unknown alias so the setter
    can fail pre-I/O before any wire bytes go out.

    Raw integer device codes (15, 30) are **not** accepted — callers
    who want the lower-level path use
    ``write_parameter("display_units", 30)``.
    """
    if isinstance(value, Unit):
        return value
    if not isinstance(value, str):
        raise WatlowValidationError(
            f"unit must be a Unit or string alias, got {type(value).__name__}",
        )
    alias = value.strip().lower()
    try:
        return _UNIT_STRING_ALIASES[alias]
    except KeyError as exc:
        raise WatlowValidationError(
            f"unknown unit alias: {value!r}",
        ) from exc

display_code_for_unit ¶

display_code_for_unit(unit)

Return the raw 17050 device code for a temperature display unit.

Source code in src/watlowlib/registry/units.py

def display_code_for_unit(unit: Unit) -> int | None:
    """Return the raw 17050 device code for a temperature display unit."""
    return _DISPLAY_CODE_FOR_UNIT.get(unit)

resolve_unit ¶

resolve_unit(kind, temperature_unit)

Resolve a parameter's :class:UnitKind to a concrete :class:Unit.

TEMPERATURE → temperature_unit (passes the caller's asserted wire scale through, or None when none was asserted).
PERCENT → :attr:Unit.PERCENT.
Everything else → None.

Pure mapping; no I/O. The caller (typically :class:watlowlib.devices.session.Session) is responsible for determining the wire scale and passing it in.

Source code in src/watlowlib/registry/units.py

def resolve_unit(kind: UnitKind, temperature_unit: Unit | None) -> Unit | None:
    """Resolve a parameter's :class:`UnitKind` to a concrete :class:`Unit`.

    - ``TEMPERATURE`` → ``temperature_unit`` (passes the caller's
      asserted wire scale through, or ``None`` when none was asserted).
    - ``PERCENT`` → :attr:`Unit.PERCENT`.
    - Everything else → ``None``.

    Pure mapping; no I/O. The caller (typically
    :class:`watlowlib.devices.session.Session`) is responsible for
    determining the wire scale and passing it in.
    """
    if kind is UnitKind.TEMPERATURE:
        return temperature_unit
    if kind is UnitKind.PERCENT:
        return Unit.PERCENT
    return None

unit_from_display_code ¶

unit_from_display_code(code)

Return the display unit for a raw 17050 device code, if known.

Source code in src/watlowlib/registry/units.py

def unit_from_display_code(code: int) -> Unit | None:
    """Return the display unit for a raw 17050 device code, if known."""
    return _DISPLAY_UNIT_CODES.get(code)

watlowlib.registry¶

Public surface¶

watlowlib.registry ¶

ControllerFamily ¶

ParameterRegistry ¶

has ¶

resolve ¶

validate_instance ¶

validate_value ¶

ParameterSpec dataclass ¶

scale class-attribute instance-attribute ¶

RwesFlag ¶

Unit ¶

UnitKind ¶

classify_family ¶

coerce_unit ¶

display_code_for_unit ¶

load_enumerations ¶

load_parameters ¶

load_pm_parameters ¶

load_sd_parameters ¶

resolve_unit ¶

unit_from_display_code ¶

Parameter specs + registry¶

watlowlib.registry.parameters ¶

ParameterRegistry ¶

has ¶

resolve ¶

validate_instance ¶

validate_value ¶

ParameterSpec dataclass ¶

scale class-attribute instance-attribute ¶

RwesFlag ¶

load_parameters ¶

load_pm_parameters ¶

load_sd_parameters ¶

Family classification + part-number decoder¶

watlowlib.registry.families ¶

ControllerFamily ¶

capabilities_for_part_number ¶

classify_family ¶

decode_part_number ¶

default_loops ¶

pm_comms_code ¶

pm_comms_supports_modbus ¶

Enumerations¶

watlowlib.registry.enumerations ¶

load_enumerations ¶

Aliases¶

watlowlib.registry.aliases ¶

Units¶

watlowlib.registry.units ¶

Unit ¶

UnitKind ¶

coerce_unit ¶

display_code_for_unit ¶

resolve_unit ¶

unit_from_display_code ¶

`watlowlib.registry`¶

ParameterSpec `dataclass` ¶

scale `class-attribute` `instance-attribute` ¶

ParameterSpec `dataclass` ¶

scale `class-attribute` `instance-attribute` ¶