"""
Category components
===================
`Resistor`, `Capacitor`, `PolarizedCapacitor` and `Inductor`: ordinary
:py:class:`jitx.Component` subclasses that resolve themselves from the active
`PartSource`. Their ports exist as soon as they are constructed -- `p1`/`p2`, or
`a`/`c` for `PolarizedCapacitor` -- so one can be wired up immediately::
self.top = Resistor(resistance=10 * kohm, tolerance=AtMost(1 * pct))
self.nets = [self.top.p2 + self.bottom.p1]
Each takes its category's query keywords, or a query object, or both; a query set
as a context in an enclosing class body supplies the defaults, and a keyword the
category does not name is passed through to the providers.
Constructing one enqueues a query rather than resolving it. The batch resolves
when something reads resolved state -- `part`, `value`, `mpn`, `manufacturer`,
`landpattern`, `symbol`, or any field a provider attached -- when the object is
introspected, or at the end of elaboration, whichever comes first. Reading early
is allowed and costs only the batching.
Two rules the whole module follows. Ports are declared eagerly, because arity is a
property of the category and not of the part. And the component does not build
geometry: it delegates to the provider's `materialize` and validates the result on
its own terms.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, ClassVar
import jitx
import jitx.units
from .error import (
PolarityMismatch,
ProviderContractError,
ResolverLostQuery,
UnresolvedPart,
)
from .query import (
CapacitorQuery,
InductorQuery,
PartQuery,
Predicate,
ResistorQuery,
require_dimension,
)
from .record import PartIdentity, ResolvedPart, apply_fields
from .resolver import DeferredState
if TYPE_CHECKING:
from pint.facets.plain import PlainUnit
[docs]
class Deferred[T]:
"""Descriptor for a component field that only exists once a part is resolved.
Reading one resolves the batch and hands back what was bound; a field
resolution never set raises `AttributeError` naming the part that was chosen,
rather than returning None. Declare one per field a category promises,
annotated with the descriptor type so a checker sees the resolved type::
value: Deferred[jitx.units.PlainQuantity] = Deferred()
"""
# A data descriptor takes precedence over both inherited class attributes and
# the instance `__dict__`, which is the whole point: `jitx.Component` declares
# `value`, `mpn` and `manufacturer` as class attributes defaulting to None, so
# an absence-based trigger would never fire for them and `r.value` would
# quietly return None instead of resolving.
_name: str
def __set_name__(self, owner: type, name: str) -> None:
self._name = name
def __get__(self, obj: Any, owner: type | None = None) -> Any:
if obj is None:
return self # class access: no resolution
ensure_resolved(obj)
try:
return obj.__dict__[self._name]
except KeyError:
raise AttributeError(self._unset_message(obj)) from None
def __set__(self, obj: Any, value: T) -> None:
# Writes the instance dict directly, which bypasses whatever
# Structurable.__setattr__ would do. These particular fields are benign,
# but it is a shortcut: "is a data descriptor storing into a Component's
# __dict__ supported?" is an open question for the JITX maintainers.
obj.__dict__[self._name] = value
def _unset_message(self, obj: Any) -> str:
# Only reached on an owner without `DeferredPart.__getattr__`. Python's
# attribute protocol treats an `AttributeError` from a descriptor as "not
# found" and falls through to `__getattr__`, which every owner of this
# descriptor has and which re-raises with its own wording. Kept because
# the descriptor must still say something sane on its own.
# `part` itself may be the field that was never set, so do not assume it.
if "part" not in obj.__dict__:
return (
f"{type(obj).__name__} was resolved but nothing was bound; "
f"{self._name!r} is unset and so is 'part'"
)
mpn = obj.__dict__["part"].identity.mpn
return (
f"{type(obj).__name__} resolved to {mpn!r} but {self._name!r} was never set"
)
[docs]
def ensure_resolved(obj: DeferredPart) -> None:
"""Flush the batch this component belongs to, unless it is already resolved.
Reentrant-safe: `_bind` writes fields and provider code reads them back while
the part is not yet marked resolved, so without the `binding` guard any such
read would re-enter the flush.
"""
state = obj._parts
if state is None or state.resolved or state.binding:
return
state.batch.flush()
[docs]
class DeferredPart:
"""Defer part resolution until something reads resolved state.
Resolution happens once and is never invalidated.
"""
# Declared with a default so the attribute is always present: absent, it would
# fall through to `__getattr__` below.
_parts: DeferredState | None = None
"""Everything this component's resolution depends on, or None before `_defer`."""
def _defer(self, query: PartQuery) -> None:
"""Enqueue `query` for this component, capturing everything it depends on."""
# The chain is captured now rather than read at flush time: a
# `with PartSource([...])` block has exited by the end of elaboration.
from .lock import Lock
from .provider import PartSource
from .resolver import PartsResolved, caller_site, enqueue
self._parts = enqueue(
self,
query,
source=PartSource.require(),
lock=Lock.require(), # never raises; Lock overrides get()
origin=caller_site(self),
)
# Registering here captures this component's frame; firing the event later
# restores it, so `materialize` runs where the component was constructed.
PartsResolved.on(self._bind_from_event)
def _bind_from_event(self, _evt: Any = None) -> None:
"""Bind the resolved part, then deregister."""
# Deregistration is required, not tidiness: handlers accumulate in
# `EventContext` for the whole run and every fire invokes all of them.
from .resolver import PartsResolved
state = self._parts
if state is None or state.part is None:
# Not this batch, or resolution failed. Leave the handler registered:
# a later flush may still be ours.
return
state.binding = True
try:
self._bind(state.part)
finally:
state.binding = False
state.resolved = True
PartsResolved.off(self._bind_from_event)
def _bind(self, part: ResolvedPart) -> None: # pragma: no cover - overridden
raise NotImplementedError
# Catch-all for anything a provider attached that the core does not name.
def __getattr__(self, name: str) -> Any:
# Underscore names raise immediately: a private lookup should never cost a
# flush. `_parts` itself never reaches here, since it is a class attribute.
if name.startswith("_"):
raise AttributeError(name)
state = self._parts
if state is None or state.binding:
# Nothing deferred, or provider code reading fields back mid-bind.
raise AttributeError(name)
if state.resolved:
# Already resolved, so the name is genuinely absent. Raise rather than
# retry: a retry here is what would recurse, and there is nothing left
# to flush that might produce the name.
raise AttributeError(self._missing(name))
state.batch.flush() # resolves the WHOLE batch
# The flush binds through `PartsResolved`, which sets `resolved`. Check it
# rather than assume it: an unresolved state here means the batch had
# nothing for us, and the retry below would not terminate.
if not state.resolved:
raise self._unavailable(name, state)
# Plain getattr, deliberately: it goes through the real lookup path rather
# than the base implementation, so an attribute a provider made visible
# some other way is still found. Re-entering `__getattr__` is harmless
# because `resolved` is now True, so the branch above turns the second
# entry into an AttributeError. One extra frame, bounded by an invariant
# just checked, rather than by bypassing the protocol.
try:
return getattr(self, name)
except AttributeError:
raise AttributeError(self._missing(name)) from None
def _unavailable(self, name: str, state: DeferredState) -> AttributeError:
"""The error for a read of a component whose query will never resolve.
Reached after a flush that left the state unresolved, which is terminal:
the flush drained `pending`, so nothing will ask this query again. Every
field of such a component raises this, `part` included, and the name that
was read is incidental -- so the message leads with the query rather than
the name, and the two failure classes are kept apart because their
remedies are unrelated. Built rather than raised so the `raise` stays at
the call site, where a traceback wants it.
"""
component = type(self).__name__
description = state.query.describe()
if state.failure is None:
return ResolverLostQuery(component, name, description, state.origin)
return UnresolvedPart(component, name, description, state.failure, state.origin)
def _missing(self, name: str) -> str:
"""Message for a name that resolution did not provide.
Reads the part off the state rather than through the `part` descriptor,
which would resolve, and which is being called from inside a failed lookup.
"""
state = self._parts
part = state.part if state is not None else None
if part is None:
return f"{type(self).__name__} has no {name!r}"
return (
f"{type(self).__name__} resolved to {part.identity.mpn!r}, "
f"which has no {name!r}"
)
def __dir__(self):
# Introspection, the translator's traversal included, must enumerate a
# resolved part: provider-attached names do not appear until then. Must
# delegate to super(), since Structurable.__dir__ filters names.
state = self._parts
if state is not None and not (state.resolved or state.binding):
state.batch.flush()
return super().__dir__()
def __repr__(self):
# Never resolve just to print. Debuggers must not collapse the batch.
state = self._parts
if state is None:
return f"<{type(self).__name__} uninitialized>"
if state.resolved:
return f"<{type(self).__name__} {self.__dict__['part'].identity.mpn}>"
return f"<{type(self).__name__} unresolved {state.query.describe()}>"
class _Passive(DeferredPart, jitx.Component):
"""Shared binding for the two-terminal categories.
Subclasses declare their own ports and deferred fields, since the set of
fields resolution provides should be visible per category rather than
inherited invisibly.
"""
query_type: ClassVar[type[PartQuery]] = PartQuery
_value_unit: ClassVar[Any] = None
def _bind(self, part: ResolvedPart) -> None:
identity = part.identity
key = self.query_type.value_key
if key is None: # pragma: no cover - categories always set one
raise ProviderContractError(
f"{type(self).__name__} has no value_key to bind"
)
try:
value = identity.attributes[key]
except KeyError:
raise ProviderContractError(
f"{identity.provider} returned {identity.mpn!r} for a "
f"{identity.category} without the required {key!r} attribute"
) from None
# The value carries its own unit, so this checks a dimension rather than
# guessing at an order of magnitude.
value = require_dimension(
value,
self._value_unit,
provider=identity.provider,
mpn=identity.mpn,
attribute=key,
)
self._check_category(identity)
self.part = part
self.value = value
apply_metadata(self, identity)
# The provider attaches whatever landpattern and symbol it sees fit.
part.materialize(self)
def _check_category(self, identity: PartIdentity) -> None:
"""Category-specific refusals. Runs before `materialize`, deliberately."""
@classmethod
def _with_unit(cls, kwargs: dict[str, object]) -> dict[str, object]:
"""`kwargs` with bare numbers in the category's value given its unit.
`Resistor(resistance=0)` reads as zero ohm, not a dimensionless zero
that no catalogue value can be compared with, and so does every bare
number in `AtMost(10)`, `(10, 22)` or `Tolerance.percent(10000, 5)`.
"""
key = cls.query_type.value_key
if key is None or key not in kwargs:
return kwargs
return {**kwargs, key: _in_unit(kwargs[key], cls._value_unit)}
def _in_unit(value: object, unit: PlainUnit) -> object:
"""`value` with every bare number in it taken in `unit`.
Takes a query value as a constructor keyword accepts it: a scalar, a tuple
or list of them, or a predicate. Anything else, a quantity included, is
returned unchanged.
"""
if isinstance(value, Predicate):
return value._apply_unit_to_bare(unit)
if isinstance(value, (tuple, list)):
return tuple(_in_unit(v, unit) for v in value)
if isinstance(value, bool) or not isinstance(value, (int, float)):
return value
return value * unit
[docs]
class Resistor(_Passive):
"""Resistor resolved from the active PartSource.
A bare number for `resistance` is taken as ohms.
"""
query_type: ClassVar[type[PartQuery]] = ResistorQuery
_value_unit: ClassVar[Any] = jitx.units.ohm
# Declared eagerly: arity is a property of the category, not of the part.
p1: jitx.Port
p2: jitx.Port
# Deferred, via data descriptors. NOT bare annotations: `value` and `mpn` are
# inherited class attributes on jitx.Component defaulting to None, so an
# absence-based trigger would never fire for them.
#
# Annotated with the descriptor type, which is not cosmetic. Written as a bare
# `mpn = Deferred[str]()` a checker reads it as assigning to the inherited
# `mpn: str | None`, which is both an error and wrong: it types `r.mpn` as
# `str | None`, the very None-returning read the descriptors exist to prevent.
# Declared this way, access goes through `__get__` and `r.mpn` is `str`.
part: Deferred[ResolvedPart] = Deferred()
value: Deferred[jitx.units.PlainQuantity] = Deferred()
landpattern: Deferred[jitx.Landpattern] = Deferred()
symbol: Deferred[jitx.Symbol] = Deferred()
# A data descriptor deliberately overriding a plain mutable attribute on
# `jitx.Component`; pyright compares declared types, not descriptor access
# types, so it cannot see that a read yields `str`.
mpn: Deferred[str] = Deferred() # pyright: ignore[reportIncompatibleVariableOverride]
manufacturer: Deferred[str | None] = Deferred() # pyright: ignore[reportIncompatibleVariableOverride]
reference_designator_prefix = "R"
def __init__(self, query: ResistorQuery | None = None, **kwargs: object):
self.p1 = jitx.Port()
self.p2 = jitx.Port()
self._defer(ResistorQuery.resolve_context(query, **self._with_unit(kwargs)))
[docs]
class Capacitor(_Passive):
"""Capacitor resolved from the active PartSource.
A bare number for `capacitance` is taken as farads.
Leaves `polarized` unset in the query, and refuses a part marked polarized at
bind time instead. A filter cannot express what this category wants: a missing
key never matches, so `polarized=Exactly(False)` would return nothing from
every catalogue that does not carry the attribute, and "unpolarized or
unmarked" has no spelling in the predicate vocabulary.
"""
query_type: ClassVar[type[PartQuery]] = CapacitorQuery
_value_unit: ClassVar[Any] = jitx.units.F
p1: jitx.Port
p2: jitx.Port
part: Deferred[ResolvedPart] = Deferred()
value: Deferred[jitx.units.PlainQuantity] = Deferred()
landpattern: Deferred[jitx.Landpattern] = Deferred()
symbol: Deferred[jitx.Symbol] = Deferred()
# A data descriptor deliberately overriding a plain mutable attribute on
# `jitx.Component`; pyright compares declared types, not descriptor access
# types, so it cannot see that a read yields `str`.
mpn: Deferred[str] = Deferred() # pyright: ignore[reportIncompatibleVariableOverride]
manufacturer: Deferred[str | None] = Deferred() # pyright: ignore[reportIncompatibleVariableOverride]
reference_designator_prefix = "C"
def __init__(self, query: CapacitorQuery | None = None, **kwargs: object):
self.p1 = jitx.Port()
self.p2 = jitx.Port()
self._defer(CapacitorQuery.resolve_context(query, **self._with_unit(kwargs)))
def _check_category(self, identity: PartIdentity) -> None:
"""Refuse a polarized part, before `materialize` attaches its geometry.
After it, the a/c pad geometry is already on a component whose ports are
p1/p2, which is the case this exists to prevent.
A missing attribute passes, which keeps the permissive default that makes
plain catalogues usable. A present non-bool is a contract error: `is True`
is the right test against a provider in contract, but a truthy non-bool
would slip through it while the provider's own `bool()` test still selected
polarized geometry.
"""
marker = identity.attributes.get("polarized")
if marker is None:
return
if not isinstance(marker, bool):
raise ProviderContractError(
f"{identity.provider} returned {identity.mpn!r} with "
f"polarized={marker!r}; the attribute vocabulary says bool, and "
f"truthiness is never consulted here"
)
if marker:
raise PolarityMismatch(identity.mpn, identity.provider)
[docs]
class PolarizedCapacitor(_Passive):
"""Capacitor with a defined polarity.
Restricts the search to polarized parts and exposes anode/cathode rather than
interchangeable pins. A sibling of `Capacitor`, not a subclass: the port names
differ, so nothing is inherited usefully and an `isinstance` relationship
between them would be a lie.
A bare number for `capacitance` is taken as farads.
"""
query_type: ClassVar[type[PartQuery]] = CapacitorQuery
_value_unit: ClassVar[Any] = jitx.units.F
a: jitx.Port # anode
c: jitx.Port # cathode
part: Deferred[ResolvedPart] = Deferred()
value: Deferred[jitx.units.PlainQuantity] = Deferred()
landpattern: Deferred[jitx.Landpattern] = Deferred()
symbol: Deferred[jitx.Symbol] = Deferred()
# A data descriptor deliberately overriding a plain mutable attribute on
# `jitx.Component`; pyright compares declared types, not descriptor access
# types, so it cannot see that a read yields `str`.
mpn: Deferred[str] = Deferred() # pyright: ignore[reportIncompatibleVariableOverride]
manufacturer: Deferred[str | None] = Deferred() # pyright: ignore[reportIncompatibleVariableOverride]
def __init__(self, query: CapacitorQuery | None = None, **kwargs: object):
self.a = jitx.Port()
self.c = jitx.Port()
# Polarity is not optional here: it narrows the search rather than
# describing the result, so it is forced into the query.
query_ = CapacitorQuery.resolve_context(query, **self._with_unit(kwargs))
self._defer(query_.update(polarized=True))
[docs]
class Inductor(_Passive):
"""Inductor resolved from the active PartSource.
A bare number for `inductance` is taken as henries.
"""
query_type: ClassVar[type[PartQuery]] = InductorQuery
_value_unit: ClassVar[Any] = jitx.units.H
p1: jitx.Port
p2: jitx.Port
part: Deferred[ResolvedPart] = Deferred()
value: Deferred[jitx.units.PlainQuantity] = Deferred()
landpattern: Deferred[jitx.Landpattern] = Deferred()
symbol: Deferred[jitx.Symbol] = Deferred()
# A data descriptor deliberately overriding a plain mutable attribute on
# `jitx.Component`; pyright compares declared types, not descriptor access
# types, so it cannot see that a read yields `str`.
mpn: Deferred[str] = Deferred() # pyright: ignore[reportIncompatibleVariableOverride]
manufacturer: Deferred[str | None] = Deferred() # pyright: ignore[reportIncompatibleVariableOverride]
reference_designator_prefix = "L"
def __init__(self, query: InductorQuery | None = None, **kwargs: object):
self.p1 = jitx.Port()
self.p2 = jitx.Port()
self._defer(InductorQuery.resolve_context(query, **self._with_unit(kwargs)))
