"""
Queries and predicates
======================
A query says what a part has to do. :py:class:`ResistorQuery`,
:py:class:`CapacitorQuery` and :py:class:`InductorQuery` carry the typed fields
for their category; the category components take the same keywords, so a query
object is only needed to share one.
Field values are predicates -- :py:class:`Exactly`, :py:class:`OneOf`,
:py:class:`Range` with its :py:func:`AtLeast` / :py:func:`AtMost` shorthands,
:py:class:`Tolerance`, :py:class:`Exists` -- and bare values and tuples coerce, so
the simple cases stay simple::
Resistor(resistance=10 * kohm, case=("0402", "0603"))
Resistor(resistance=Range(9 * kohm, 11 * kohm), tolerance=AtMost(0.01))
``rank=`` orders the matches rather than filtering them, applied
lexicographically, and a bare key name means :py:func:`Minimize`. A keyword the
category does not name lands in ``extras`` and reaches providers as written, so a
provider-specific attribute needs no ceremony -- in the constructor as much as in
:py:meth:`PartQuery.update`, which returns a modified copy.
A query is an immutable, hashable :py:class:`jitx.context.Context`, so one assigned
in a ``Design`` or ``Circuit`` class body supplies the defaults for everything
constructed below it::
class Board(Design):
shared = ResistorQuery(case="0402", aec_q200=True)
That is the spelling the untyped keys mostly serve: a shared query carries house
policy, while the value of a part is per instance.
Dimensioned values are :py:class:`~jitx.units.PlainQuantity` everywhere and keep
their units all the way to the provider. Only :py:meth:`PartQuery.digest` reduces
them, so that ``10 * kohm`` and ``10000 * ohm`` are one query and share one
lockfile entry.
:py:func:`match`, :py:func:`satisfies` and :py:func:`rank_by` are where
predicate-against-value semantics are defined. A provider uses them rather than
reimplementing them, which is why they are re-exported from
:py:mod:`jitxlib.parts.provider`.
"""
from __future__ import annotations
import hashlib
import json
import math
from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import dataclass, fields, replace
from typing import (
TYPE_CHECKING,
Any,
ClassVar,
Literal,
Final,
Self,
TypeGuard,
TypeVar,
cast,
get_origin,
)
import jitx.context
from jitx.units import PlainQuantity, registry
from .error import DimensionMismatch
if TYPE_CHECKING:
from pint.facets.plain import PlainUnit
Scalar = float | int | str | bool | PlainQuantity
# The subset arithmetic is defined on. `Tolerance` needs it: a percentage of a
# string is not a thing, and typing its fields as `Scalar` only hid that.
Numeric = float | int | PlainQuantity
T = TypeVar("T")
S = TypeVar("S", bound=Scalar | None)
def _as_scalar(value: object) -> Scalar:
"""Check a query value is something a predicate can hold, and return it.
A real check rather than a cast: a query value of the wrong shape should say
so here, where it was written, not fail obscurely at match time.
"""
if isinstance(value, (bool, int, float, str, PlainQuantity)):
return value
raise TypeError(
f"query values must be a number, a quantity, a string or a bool, "
f"not {type(value).__name__}"
)
[docs]
def is_quantity(value: object) -> TypeGuard[PlainQuantity]:
"""Narrowing helper. `PlainQuantity` is generic, so a bare isinstance does
not narrow for a type checker."""
return isinstance(value, PlainQuantity)
# Fields that live on PartIdentity rather than in PartIdentity.attributes. A
# filter naming one of these is matched against the identity.
IDENTITY_KEYS = frozenset({"mpn", "manufacturer", "category"})
# Query fields that are not filters: they steer the machinery instead.
_NON_FILTER_FIELDS = frozenset({"extras", "rank", "lock_scope"})
[docs]
@dataclass(frozen=True)
class Predicate:
"""Base for query value constraints. Use one of the subclasses below, or pass a
bare value or tuple and let :py:func:`coerce` pick one."""
# A dataclass itself, so `fields()` accepts any predicate: every subclass is a
# frozen dataclass, and saying so lets the encoders introspect one without a
# cast.
def _apply_unit_to_bare(self, unit: PlainUnit) -> Self:
"""A copy with every bare number it holds taken in `unit`.
Quantities, strings and bools are left as they are, so a predicate
already written with units comes back equal.
"""
return self
@staticmethod
def _unit_if_bare(value: S, unit: PlainUnit) -> S | PlainQuantity:
"""`value` in `unit` if it is a bare number, else `value` unchanged."""
if isinstance(value, bool) or not isinstance(value, (int, float)):
return value
return value * unit
[docs]
@dataclass(frozen=True)
class Exactly(Predicate):
"""Matches a value equal to `value`."""
value: Scalar
def _apply_unit_to_bare(self, unit: PlainUnit) -> Self:
return replace(self, value=self._unit_if_bare(self.value, unit))
[docs]
@dataclass(frozen=True)
class OneOf(Predicate):
"""Matches a value equal to any member of `values`."""
values: tuple[Scalar, ...]
def _apply_unit_to_bare(self, unit: PlainUnit) -> Self:
return replace(
self, values=tuple(self._unit_if_bare(v, unit) for v in self.values)
)
[docs]
@dataclass(frozen=True)
class Range(Predicate):
"""Closed interval; either bound may be None for open-ended.
Bounds are quantities for dimensioned attributes, plain numbers for
dimensionless ones, and must not be mixed within one Range.
"""
min: Scalar | None = None
max: Scalar | None = None
def _apply_unit_to_bare(self, unit: PlainUnit) -> Self:
return replace(
self,
min=self._unit_if_bare(self.min, unit),
max=self._unit_if_bare(self.max, unit),
)
[docs]
def AtLeast(v: Scalar) -> Range:
"""Range with only a lower bound."""
return Range(min=v)
[docs]
def AtMost(v: Scalar) -> Range:
"""Range with only an upper bound."""
return Range(max=v)
[docs]
@dataclass(frozen=True)
class Tolerance(Predicate):
"""Typical value with asymmetric bounds, mirroring :py:class:`jitx.Toleranced`.
`plus` and `minus` are increments from `typ` to the maximum and minimum, not
relative fractions, and None means unbounded on that side. Unlike
`Toleranced`, all three may be quantities.
>>> Tolerance(10 * kohm, 500 * ohm, 500 * ohm) # 10k +/- 500R
>>> Tolerance.percent(10 * kohm, 5) # 10k +/- 5%
>>> Tolerance(10 * kohm, None, 500 * ohm) # unbounded above
"""
typ: Numeric
plus: Numeric | None
minus: Numeric | None
def _apply_unit_to_bare(self, unit: PlainUnit) -> Self:
return replace(
self,
typ=self._unit_if_bare(self.typ, unit),
plus=self._unit_if_bare(self.plus, unit),
minus=self._unit_if_bare(self.minus, unit),
)
[docs]
@classmethod
def percent(
cls, typ: Numeric, plus: float, minus: float | None = None
) -> Tolerance:
"""Relative form, asymmetric like `Toleranced.percent`.
`minus` defaults to `plus`. The percentages are dimensionless whatever
`typ` is.
"""
if minus is None:
minus = plus
return cls(typ, typ * plus / 100, typ * minus / 100)
[docs]
@dataclass(frozen=True)
class Exists(Predicate):
"""Matches on presence (or, with `present=False`, absence) of the key."""
present: bool = True
[docs]
def coerce(value: object) -> Predicate:
"""Turn a bare value into a predicate.
Scalars become `Exactly`, tuples and lists become `OneOf`, and a predicate
passes through unchanged.
"""
if isinstance(value, Predicate):
return value
if isinstance(value, (tuple, list)):
return OneOf(tuple(_as_scalar(v) for v in value))
return Exactly(_as_scalar(value))
def _as_extras(value: object) -> dict[str, Predicate]:
"""Read an `extras=` argument in either accepted form, coercing the values."""
items: Iterable[object]
if isinstance(value, Mapping):
items = cast(Mapping[object, object], value).items()
elif isinstance(value, (tuple, list)):
items = cast(Sequence[object], value)
else:
raise TypeError(
f"extras must be a mapping of key to predicate, or pairs, "
f"not {type(value).__name__}"
)
out: dict[str, Predicate] = {}
for item in items:
if (
not isinstance(item, (tuple, list))
or len(cast(Sequence[object], item)) != 2
):
raise TypeError("each extras entry must be a (key, value) pair")
key, val = cast(Sequence[object], item)
out[str(key)] = coerce(val)
return out
[docs]
@dataclass(frozen=True)
class Rank:
"""Order results by `key`. Not a filter; never excludes a part."""
key: str
direction: Literal["min", "max"] = "min"
[docs]
def Minimize(key: str) -> Rank:
"""Rank ascending by `key`."""
return Rank(key, "min")
[docs]
def Maximize(key: str) -> Rank:
"""Rank descending by `key`."""
return Rank(key, "max")
def _as_ranks(value: object) -> Rank | str | Iterable[Rank | str]:
"""Check a `rank=` argument is something `_coerce_rank` can walk.
Same reasoning as `_as_scalar`: the error belongs where the argument was
written, not three frames further in.
"""
if isinstance(value, (Rank, str)) or isinstance(value, Iterable):
return value
raise TypeError(
f"rank must be a Rank, a key name, or an iterable of either, "
f"not {type(value).__name__}"
)
# What a query field accepts, as opposed to what it holds. A bare value or tuple
# is coerced at construction, so after that every filter is a `Predicate` and
# every rank a `Rank`, but the declared type has to describe the input, since
# that is what a caller writes.
type Filter = Predicate | Scalar | tuple[Scalar, ...] | list[Scalar]
type RankSpec = Rank | str | tuple[Rank | str, ...] | list[Rank | str]
# `extras` accepts the keyword mapping a caller would write and the pairs form
# it is stored as, since a copy round-trips through the constructor.
type ExtrasSpec = Mapping[str, Filter] | tuple[tuple[str, Filter], ...]
def _coerce_rank(value: Rank | str | Iterable[Rank | str]) -> tuple[Rank, ...]:
"""Accept a Rank, a bare key string, or an iterable of either."""
if isinstance(value, Rank):
return (value,)
if isinstance(value, str):
return (Minimize(value),)
out: list[Rank] = []
for item in value:
out.extend(_coerce_rank(item))
return tuple(out)
# --------------------------------------------------------------------------
# Comparison, where predicate-against-value semantics are defined.
# --------------------------------------------------------------------------
def _comparable(a: object, b: object, key: str) -> tuple[Any, Any]:
"""Put two values on comparable footing, or raise.
A dimensionless quantity and a bare number denote the same thing, so the
quantity is reduced to its magnitude. A genuine dimensionality mismatch is a
bug rather than a non-match, and the error names both dimensions without
presuming whose bug it is.
"""
# Inline guards, not `aq = is_quantity(a)`: assigning a TypeGuard's result to a
# variable narrows nothing, which is what the casts here were papering over.
if is_quantity(a):
if is_quantity(b):
if a.dimensionality != b.dimensionality:
raise DimensionMismatch(key, a, b)
return a, b
if a.dimensionless:
# to_base_units(), not .magnitude: (1 * pct).magnitude is 1, and the
# thing it denotes is 0.01.
return a.to_base_units().magnitude, b
if isinstance(b, (int, float)) and not isinstance(b, bool):
raise DimensionMismatch(key, a, b)
return a, b
if is_quantity(b):
other, _ = _comparable(b, a, key)
return (a, other) if other is not b else (a, b)
return a, b
# Component values are physical quantities, not IEEE doubles, and a unit
# conversion does not round-trip exactly: `100 * nF` converts to a farad
# magnitude that is not bit-identical to `1e-7`, so `==` says they differ. Nobody
# writing `Exactly(100 * nF)` means "and not the same value spelled in farads",
# so numeric comparison is relative-tolerance throughout. The tolerance is far
# tighter than any real component tolerance, so it cannot merge distinct values.
_REL_TOL = 1e-9
# Significant digits the digest keeps. Same problem one level up: two spellings of
# one value must hash alike, and their base-unit magnitudes differ in the last
# bits.
_DIGEST_SIG_DIGITS = 12
def _magnitude(value: object) -> float | None:
"""Base-unit magnitude of a number or quantity, or None if it is neither."""
if isinstance(value, bool):
return None
if is_quantity(value):
return float(value.to_base_units().magnitude)
if isinstance(value, (int, float)):
return float(value)
return None
def _close(a: object, b: object) -> bool | None:
"""Whether two numeric values are equal to within `_REL_TOL`.
None when either side is not numeric, so the caller falls back to `==`.
"""
x, y = _magnitude(a), _magnitude(b)
if x is None or y is None:
return None
return math.isclose(x, y, rel_tol=_REL_TOL, abs_tol=0.0)
def _eq(value: object, bound: object, key: str) -> bool:
a, b = _comparable(value, bound, key)
near = _close(a, b)
if near is not None:
return near
try:
return bool(a == b)
except (TypeError, ValueError):
return False
def _le(a: object, b: object, key: str) -> bool:
x, y = _comparable(a, b, key)
# A bound is inclusive, so a value sitting exactly on it must satisfy it --
# which `<=` alone gets wrong when the two came from different unit
# spellings.
if _close(x, y):
return True
try:
return bool(x <= y)
except (TypeError, ValueError) as exc:
raise DimensionMismatch(key, a, b) from exc
[docs]
def satisfies(predicate: Predicate | None, value: object, key: str = "") -> bool:
"""Whether one value satisfies one predicate.
`predicate` of None is no constraint and always matches. Presence of the key
is the caller's business, except for `Exists`, which is about exactly that;
see :py:func:`match`.
"""
if predicate is None:
return True
if isinstance(predicate, Exists):
return predicate.present
if isinstance(predicate, Exactly):
return _eq(value, predicate.value, key)
if isinstance(predicate, OneOf):
return any(_eq(value, v, key) for v in predicate.values)
if isinstance(predicate, Range):
if predicate.min is not None and not _le(predicate.min, value, key):
return False
if predicate.max is not None and not _le(value, predicate.max, key):
return False
return True
if isinstance(predicate, Tolerance):
typ = predicate.typ
if predicate.minus is not None:
if not _le(typ - predicate.minus, value, key):
return False
if predicate.plus is not None:
if not _le(value, typ + predicate.plus, key):
return False
if predicate.plus is None and predicate.minus is None:
return _eq(value, typ, key)
return True
raise TypeError(f"unknown predicate type {type(predicate).__name__}")
[docs]
def satisfies_or_true(
predicate: Predicate | None, value: object, key: str = ""
) -> bool:
"""`satisfies`, but a None predicate is no constraint.
What a generator provider wants: it synthesized a value and needs to know
whether the query would have accepted it, with unconstrained keys passing.
"""
return True if predicate is None else satisfies(predicate, value, key)
[docs]
def match(
filters: Mapping[str, Predicate],
attributes: Mapping[str, Scalar],
identity: object | None = None,
) -> bool:
"""Whether a candidate satisfies every filter.
Identity keys (`mpn`, `manufacturer`, `category`) are read from `identity`
rather than from `attributes`, since they are fields of `PartIdentity`; a
filter on one of them with `identity=None` does not match.
A missing key never matches, except `Exists(False)`. A dimensionality
mismatch raises :py:class:`DimensionMismatch` rather than returning False,
because it is a bug and not a non-match.
"""
for key, predicate in filters.items():
if predicate is None:
continue
if key in IDENTITY_KEYS:
present = identity is not None and getattr(identity, key, None) is not None
value = getattr(identity, key, None) if identity is not None else None
else:
present = key in attributes
value = attributes.get(key)
if isinstance(predicate, Exists):
if predicate.present != present:
return False
continue
if not present:
return False
if not satisfies(predicate, value, key):
return False
return True
[docs]
def rank_by(
items: Iterable[T],
ranks: Sequence[Rank],
key: Callable[[T], Mapping[str, Scalar]],
) -> list[T]:
"""Order `items` by `ranks`, applied lexicographically.
`key` reaches the attribute mapping for an item, since callers rank things
that are not themselves attribute maps. Items missing a ranked key sort
last, whichever direction is asked for: absent is not better than known.
"""
out = list(items)
for rank in reversed(list(ranks)):
reverse = rank.direction == "max"
def sort_key(item: T, _rank: Rank = rank) -> tuple[int, Any]:
value = key(item).get(_rank.key)
if value is None:
return (1, 0)
if is_quantity(value):
value = value.to_base_units().magnitude
return (0, value)
present = [i for i in out if key(i).get(rank.key) is not None]
absent = [i for i in out if key(i).get(rank.key) is None]
present.sort(key=sort_key, reverse=reverse)
out = present + absent
return out
# --------------------------------------------------------------------------
# Canonicalization and digest
# --------------------------------------------------------------------------
def _digest_value(value: object) -> Any:
"""Reduce a value to a stable, JSON-encodable form for hashing.
Quantities become their base-unit magnitude, so equivalent spellings of the
same query agree.
"""
if is_quantity(value):
magnitude = round_significant(float(value.to_base_units().magnitude))
if value.dimensionless:
# A dimensionless quantity is its magnitude, so `1 * pct` digests as
# `0.01` and agrees with the bare float. Same rule as the lockfile's
# `encode_value`, which the digest used to contradict.
return magnitude
return ["q", magnitude]
if isinstance(value, tuple):
return [_digest_value(v) for v in value]
return value
[docs]
def round_significant(value: float, digits: int = _DIGEST_SIG_DIGITS) -> float:
"""Round to `digits` significant digits, to shed float representation noise.
Two callers, same purpose. The digest needs it so that two spellings of one
value hash alike: `100 * nF` and `1e-7 * F` have base-unit magnitudes that
differ in the last bits. A provider rescaling a value for display needs it
because the rescale has the same problem in reverse: `1e-7 F` converts to
`99.99999999999999 nanofarad`, which is not what anyone wants on a schematic.
"""
if value == 0.0 or not math.isfinite(value):
return value
exponent = math.floor(math.log10(abs(value)))
return round(value, -(exponent - (_DIGEST_SIG_DIGITS - 1)))
def _digest_predicate(predicate: Predicate) -> Any:
name = type(predicate).__name__
values = [_digest_value(getattr(predicate, f.name)) for f in fields(predicate)]
return [name, *values]
[docs]
@dataclass(frozen=True)
class CanonicalQuery:
"""The flat, category-tagged form that crosses the provider boundary.
Quantities keep their units here; only `digest` reduces them. `lock_scope`
is deliberately absent, since it has no bearing on which parts match.
"""
category: str
value_key: str | None
filters: Mapping[str, Predicate]
rank: tuple[Rank, ...]
limit: int
digest: str
count: int = 1
"""How many components in this batch asked this exact question.
Demand, not a constraint: it says nothing about which parts match, takes no
part in the digest, and a provider that ignores it is correct. It is here for
the provider that cannot do its job without it -- minimum order quantities,
bulk rebates, consolidating a reel -- which are properties of the order rather
than of one query, and are not expressible from a batch that has had its
duplicates collapsed.
Two caveats, both structural. Batching is best effort, so an early flush
splits one order into several batches and no batch can be assumed to be the
whole design. And a query the lock answers never reaches a provider at all, so
on a run with a populated lock the batch is the *unpinned remainder*, not the
order -- except in `check`'s clean-run call, which carries every query the
design asked precisely so the comparison is against a whole-order resolution. An optimization whose answer depends on the whole order is therefore in
tension with pinning by nature: adding one component could legitimately move
every selection, which is what a lock exists to prevent. A provider that makes
that trade owns it.
"""
[docs]
def describe(self) -> str:
"""One-line rendering, for diagnostics and lockfile display strings."""
parts = [f"{k}={_describe_predicate(v)}" for k, v in self.filters.items()]
if self.rank:
ranks = ", ".join(f"{r.direction}({r.key})" for r in self.rank)
parts.append(f"rank=({ranks})")
return f"{self.category}({', '.join(parts)})"
def _describe_predicate(predicate: Predicate) -> str:
if isinstance(predicate, Exactly):
return f"{predicate.value}"
if isinstance(predicate, OneOf):
return "OneOf(" + ", ".join(str(v) for v in predicate.values) + ")"
if isinstance(predicate, Range):
if predicate.min is None:
return f"<={predicate.max}"
if predicate.max is None:
return f">={predicate.min}"
return f"{predicate.min}..{predicate.max}"
if isinstance(predicate, Tolerance):
return f"{predicate.typ} +{predicate.plus}/-{predicate.minus}"
if isinstance(predicate, Exists):
return "present" if predicate.present else "absent"
return repr(predicate)
# Per-class field lists, computed once. Keyed by the class, which is what the
# `__annotations__` walk is keyed on.
_FIELD_NAMES: dict[type, tuple[str, ...]] = {}
def _is_classvar(annotation: object) -> bool:
"""Whether an annotation marks a class attribute rather than a field."""
# `from __future__ import annotations` makes every annotation in this module a
# string, but a subclass defined elsewhere need not have that import.
if isinstance(annotation, str):
return annotation.lstrip().startswith(
("ClassVar", "typing.ClassVar", "t.ClassVar")
)
return get_origin(annotation) is ClassVar
# Not a dataclass. The synthesized `__init__` is a closed signature, and a query
# has to take the provider-specific keys that `update()` takes; there is no way
# to say "the declared fields plus arbitrary keywords" to `@dataclass`. What the
# decorator gave is written out below instead: the field list comes off
# `__annotations__`, which is where the dataclass read it too, and equality,
# hashing and immutability are a few lines each. The per-class `__init__`
# declarations under TYPE_CHECKING are what keeps call sites checked; a test
# guards them against drifting from the annotations.
[docs]
class PartQuery(jitx.context.Context):
"""Base query. Subclasses add typed fields for their category.
Usable as a design-context default, but never resolvable on its own: it has
no `value_key`, so no category component will accept it.
The constructor takes what `update` takes: a keyword the category does not
name lands in `extras` rather than raising.
"""
category: ClassVar[str] = "part"
value_key: ClassVar[str | None] = None
# `Final` on every field is how assignment stays an error for a type checker
# now that `frozen=True` is gone; `__setattr__` below is the runtime half. A
# custom `__setattr__` alone does not do it, since pyright only consults one
# for attributes the class has not declared.
mpn: Final[Filter | None] = None
"""Manufacturer part number. Filtering on it asks for one specific part."""
manufacturer: Final[Filter | None] = None
"""Manufacturer name, as the provider spells it."""
case: Final[Filter | None] = None
"""Package code, such as "0402" or "SOT-23-5"."""
extras: Final[tuple[tuple[str, Filter], ...]] = ()
"""Filters on attributes the core does not name, such as `lcsc` or `aec_q200`.
Pass them as loose keywords or as a mapping; this pairs form is what they are
stored as.
"""
# Pairs rather than a Mapping because a Context must be hashable; the
# constructor normalizes to sorted pairs.
rank: Final[RankSpec] = ()
"""How to order the matches, applied lexicographically.
A bare key minimizes, so `rank="price"` means `Minimize("price")`.
"""
lock_scope: Final[str] = ""
"""Pins otherwise identical queries to separate lockfile entries."""
# Not a filter, not part of the digest, never sent to a provider.
def _initialize(self, kwargs: Mapping[str, object]) -> None:
# The real constructor body. It lives here rather than in `__init__` so
# that it is type-checked: the runtime `__init__` sits in the `else` of a
# TYPE_CHECKING branch, which a checker treats as unreachable.
#
# Bare values are coerced here, not later. The dataclass version coerced
# in `update()` and not in the constructor, which made
# `ResistorQuery(resistance=10 * kohm)`, the spelling the exploration API
# and a design-level query both use, hold a quantity where a predicate
# belongs, and that failed at match time rather than here.
state = self._normalized(kwargs, ())
for name in self._field_names():
# Class-body assignments are the defaults, read the same way a
# dataclass read them.
object.__setattr__(
self, name, state.get(name, getattr(type(self), name, None))
)
if TYPE_CHECKING:
def __init__(
self,
*,
mpn: Filter | None = ...,
manufacturer: Filter | None = ...,
case: Filter | None = ...,
extras: ExtrasSpec = ...,
rank: RankSpec = ...,
lock_scope: str = ...,
**filters: Filter,
) -> None: ...
else:
def __init__(self, **kwargs):
self._initialize(kwargs)
def __setattr__(self, name: str, value: object) -> None:
raise AttributeError(
f"{type(self).__name__} is immutable; use update() for a modified copy"
)
def __delattr__(self, name: str) -> None:
raise AttributeError(f"{type(self).__name__} is immutable")
def _state(self) -> tuple[object, ...]:
return tuple(getattr(self, name, None) for name in self._field_names())
def __eq__(self, other: object) -> bool:
if type(other) is not type(self):
return NotImplemented
return self._state() == cast(PartQuery, other)._state()
def __hash__(self) -> int:
return hash((type(self), self._state()))
def __repr__(self) -> str:
# Only what was set: a query has many fields and almost all of them are
# unset, so the dataclass repr was mostly `=None`.
shown = [
f"{name}={getattr(self, name, None)!r}"
for name in self._field_names()
if getattr(self, name, None) != getattr(type(self), name, None)
]
return f"{type(self).__name__}({', '.join(shown)})"
@classmethod
def _field_names(cls) -> tuple[str, ...]:
"""Every declared field, base first. `ClassVar`s and privates excluded."""
cached = _FIELD_NAMES.get(cls)
if cached is None:
names: list[str] = []
for base in reversed(cls.__mro__):
for name, annotation in vars(base).get("__annotations__", {}).items():
if name.startswith("_") or name in names:
continue
if _is_classvar(annotation):
continue
names.append(name)
cached = tuple(names)
_FIELD_NAMES[cls] = cached
return cached
@classmethod
def _typed_filter_names(cls) -> tuple[str, ...]:
return tuple(n for n in cls._field_names() if n not in _NON_FILTER_FIELDS)
@classmethod
def _normalized(
cls, kwargs: Mapping[str, object], base_extras: tuple[tuple[str, Filter], ...]
) -> dict[str, Any]:
"""Split keywords into field values, coercing each.
Shared by the constructor and `update` so the two cannot disagree.
`base_extras` is what the result merges onto, empty when constructing.
Args:
kwargs: The keywords as written by the caller.
base_extras: Existing extras to merge into, in stored pairs form.
"""
typed = set(cls._typed_filter_names())
direct: dict[str, Any] = {}
extras: dict[str, Predicate] = {k: coerce(v) for k, v in base_extras}
# `extras=` is read before the loose keys whatever order it was written
# in: it replaces, and the loose keys then refine it. Taking it in kwargs
# order made the result depend on the spelling.
if "extras" in kwargs:
extras = _as_extras(kwargs["extras"])
for name, value in kwargs.items():
if name == "extras":
continue
if name == "rank":
direct["rank"] = () if not value else _coerce_rank(_as_ranks(value))
elif name == "lock_scope":
direct["lock_scope"] = str(value)
elif name in typed:
direct[name] = None if value is None else coerce(value)
else:
extras[name] = coerce(value)
direct["extras"] = tuple(sorted(extras.items(), key=lambda kv: kv[0]))
return direct
[docs]
@classmethod
def from_query(cls, query: PartQuery) -> Self:
"""Build a category query from a more general one.
Nothing is dropped: base fields carry over, `extras` carries over, and a
field that is not typed on the target lands in `extras` rather than being
discarded. An unserved key then surfaces at resolution time with the
aggregated diagnostic, which is the same treatment a typo gets.
"""
# The demotion is the constructor's own rule now, so this only has to
# hand everything over as keywords. An extras key that the target does
# type is promoted to that field, which `filters()` merged anyway.
kwargs: dict[str, Any] = dict(query.extras)
for name in query._typed_filter_names():
value = getattr(query, name, None)
if value is not None:
kwargs[name] = value
kwargs["rank"] = query.rank
kwargs["lock_scope"] = query.lock_scope
return cls(**kwargs)
[docs]
@classmethod
def resolve_context(cls, query: PartQuery | None = None, **kwargs: object) -> Self:
"""The `Resistor(query=None, **kwargs)` path.
Takes the explicit query if given, else the active one from the design
context, else an empty one; then applies `kwargs`.
"""
if query is None:
query = cls.get() or PartQuery.get()
if query is None:
base = cls()
elif isinstance(query, cls):
base = query
else:
base = cls.from_query(query)
out = base.update(**kwargs) if kwargs else base
return cast(Self, out)
[docs]
def update(self, **kwargs: object) -> Self:
"""Return a copy with `kwargs` applied.
Unknown keywords land in `extras`; every other field is replaced, never
merged, `rank` included.
"""
state: dict[str, Any] = {
name: getattr(self, name, None) for name in self._field_names()
}
state.update(self._normalized(kwargs, self.extras))
return type(self)(**state)
[docs]
def ranks(self) -> tuple[Rank, ...]:
"""The rank tuple, narrowed. Idempotent: construction already coerced."""
return _coerce_rank(self.rank) if self.rank else ()
[docs]
def filters(self) -> dict[str, Predicate]:
"""Typed fields and extras, merged and sorted by key."""
out: dict[str, Predicate] = {}
for name in self._typed_filter_names():
value = getattr(self, name, None)
if value is not None:
out[name] = coerce(value)
out.update((k, coerce(v)) for k, v in self.extras)
return dict(sorted(out.items(), key=lambda kv: kv[0]))
[docs]
def canonical(self, limit: int = 1) -> CanonicalQuery:
"""Flatten to the form a provider sees. Units are kept."""
filters = self.filters()
return CanonicalQuery(
category=self.category,
value_key=self.value_key,
filters=filters,
rank=self.ranks(),
limit=limit,
digest=_digest_of(self.category, self.value_key, filters, self.ranks()),
)
[docs]
def digest(self) -> str:
"""Stable hash over category, value_key, filters and rank.
Excludes `limit` (transport only), `count` (demand, not a constraint) and
`lock_scope` (a partition, not a constraint). Quantities are reduced to
base-unit magnitudes, so equivalent spellings of one query hash alike.
"""
return _digest_of(self.category, self.value_key, self.filters(), self.ranks())
def _digest_of(
category: str,
value_key: str | None,
filters: Mapping[str, Predicate],
rank: Sequence[Rank],
) -> str:
payload = [
category,
value_key,
[[k, _digest_predicate(v)] for k, v in sorted(filters.items())],
[[r.key, r.direction] for r in rank],
]
blob = json.dumps(payload, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(blob.encode("utf-8")).hexdigest()
[docs]
class PassiveQuery(PartQuery):
"""Shared by the two-terminal passive categories."""
tolerance: Final[Filter | None] = None
"""Relative tolerance, dimensionless. Write `0.01` or `1 * pct`."""
if TYPE_CHECKING:
def __init__(
self,
*,
mpn: Filter | None = ...,
manufacturer: Filter | None = ...,
case: Filter | None = ...,
extras: ExtrasSpec = ...,
rank: RankSpec = ...,
lock_scope: str = ...,
tolerance: Filter | None = ...,
**filters: Filter,
) -> None: ...
[docs]
class ResistorQuery(PassiveQuery):
"""Query for a resistor."""
category: ClassVar[str] = "resistor"
value_key = "resistance"
resistance: Final[Filter | None] = None
"""Resistance, as a quantity: `10 * kohm`."""
if TYPE_CHECKING:
def __init__(
self,
*,
mpn: Filter | None = ...,
manufacturer: Filter | None = ...,
case: Filter | None = ...,
extras: ExtrasSpec = ...,
rank: RankSpec = ...,
lock_scope: str = ...,
tolerance: Filter | None = ...,
resistance: Filter | None = ...,
**filters: Filter,
) -> None: ...
[docs]
class CapacitorQuery(PassiveQuery):
"""Query for a capacitor."""
category: ClassVar[str] = "capacitor"
value_key = "capacitance"
capacitance: Final[Filter | None] = None
"""Capacitance, as a quantity: `100 * nF`."""
rated_voltage_dc: Final[Filter | None] = None
"""Rated DC voltage, as a quantity: `AtLeast(16 * V)`."""
temperature_coefficient_code: Final[Filter | None] = None
"""Dielectric code, such as "X7R" or "C0G"."""
polarized: Final[Filter | None] = None
"""Whether the part is polarized.
`PolarizedCapacitor` sets this; `Capacitor` leaves it unset and refuses a
polarized part when it binds, so an unpolarized part needs no filter here.
"""
if TYPE_CHECKING:
def __init__(
self,
*,
mpn: Filter | None = ...,
manufacturer: Filter | None = ...,
case: Filter | None = ...,
extras: ExtrasSpec = ...,
rank: RankSpec = ...,
lock_scope: str = ...,
tolerance: Filter | None = ...,
capacitance: Filter | None = ...,
rated_voltage_dc: Filter | None = ...,
temperature_coefficient_code: Filter | None = ...,
polarized: Filter | None = ...,
**filters: Filter,
) -> None: ...
[docs]
class InductorQuery(PassiveQuery):
"""Query for an inductor."""
category: ClassVar[str] = "inductor"
value_key = "inductance"
inductance: Final[Filter | None] = None
"""Inductance, as a quantity: `4.7 * uH`."""
if TYPE_CHECKING:
def __init__(
self,
*,
mpn: Filter | None = ...,
manufacturer: Filter | None = ...,
case: Filter | None = ...,
extras: ExtrasSpec = ...,
rank: RankSpec = ...,
lock_scope: str = ...,
tolerance: Filter | None = ...,
inductance: Filter | None = ...,
**filters: Filter,
) -> None: ...
[docs]
def require_dimension(
value: object,
unit: PlainQuantity,
*,
provider: str,
mpn: str,
attribute: str,
) -> PlainQuantity:
"""Check a provider's value carries the right dimensionality, and return it.
The whole of the unit contract: there is no magnitude heuristic and no
conversion. Reported against the provider, with the part and attribute
named.
"""
from .error import ProviderContractError
if not is_quantity(value):
raise ProviderContractError(
f"{provider} returned {mpn!r} with {attribute}={value!r}, "
f"which carries no unit; expected a quantity in {unit.units}"
)
if value.dimensionality != unit.dimensionality:
raise ProviderContractError(
f"{provider} returned {mpn!r} with {attribute}={value}, "
f"expected dimensionality {unit.dimensionality} "
f"but got {value.dimensionality}"
)
return value
[docs]
def quantity(text: str) -> PlainQuantity:
"""Parse a quantity through JITX's shared unit registry.
Every quantity in this package has to come from `jitx.units.registry`:
quantities from two registries do not compare.
"""
return registry.Quantity(text)
