"""
CsvParts
========
Resolve passives from a CSV of approved parts. Name it in a chain as a path, or as
a resource inside an installed package::
[tool.jitx.parts]
chain = ["csv:parts/approved-parts.csv"]
PartSource([CsvParts("parts/approved-parts.csv")])
PartSource([CsvParts.from_package("acme_parts", "approved.csv")])
The header is the vocabulary. `category`, `mpn` and `manufacturer` are identity,
`datasheet` is reported but not queryable, and every other column becomes a
queryable attribute under its own name::
category,mpn,manufacturer,resistance,case,tolerance,price
resistor,RC0402FR-0710KL,Yageo,10000,0402,0.01,0.0032
resistor,ERA-2AEB103X,Panasonic,10 kohm,0402,0.001,0.0900
A dimensioned column takes a bare number or one carrying its own unit. A bare
number is given an engineering prefix, so it reaches the schematic as `10 kohm`
rather than `10000 ohm`; a written unit is used as written. Adding a column is all
it takes to filter or rank on it, here or in a design.
Being column-driven rather than hardcoded is also what makes this the one to read
first when writing another provider: the vocabulary comes from the data, so what
is left in the file is the protocol and nothing else.
"""
from __future__ import annotations
import csv
import functools
import io
import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from importlib import resources
from pathlib import Path
from typing import Any
from jitx.units import PlainQuantity, registry
from ..error import StalePayload
from ..provider import (
CanonicalQuery,
CategoryProvider,
NoMatch,
Outcome,
Resolved,
Unsupported,
apply_fields,
decode_payload,
encode_payload,
match,
rank_by,
)
from ..query import round_significant
from ..record import JSON, PartIdentity, ResolvedPart
# Columns this provider consumes itself rather than exposing as attributes.
# `datasheet` is in here because it is not queryable and not identity: it rides in
# the payload and is set on the component in materialize.
_RESERVED_COLUMNS = frozenset({"category", "mpn", "manufacturer", "datasheet"})
# Payload keys that are the provider's bookkeeping rather than data to report.
_PRIVATE_PAYLOAD = frozenset({"v"})
# The unit each dimensioned attribute is parsed into when the CSV gives a bare
# number. A column whose values already carry units ("10 kohm") is parsed as
# written and this table is not consulted. Not limited to typed query fields:
# `max_power`, `max_current` and `area` are common catalogue columns that reach a
# query through extras, and dropping them here would turn a bare-number column
# dimensionless. `low_esr` in `_BOOLS` stays for the same reason.
_UNITS: Mapping[str, str] = {
"resistance": "ohm",
"capacitance": "F",
"inductance": "H",
"rated_voltage_dc": "V",
"max_power": "W",
"max_current": "A",
"area": "m**2",
}
_BOOLS = frozenset({"polarized", "low_esr"})
# Vocabulary keys whose values are strings. Honouring the vocabulary beats
# guessing from the text: a case code like "1206" is not the number 1206, and
# sniffing turned it into one, which made the part invisible to a query asking
# for `case="1206"` and would have handed the integer to a landpattern generator.
# A leading number followed by something else: "1.2 mm", "50 mohm", "5%". The
# guard matters more than the parse: pint reads a bare word as a unit, so "A"
# becomes one ampere and "min" one minute, and a column of grades or codes would
# quietly turn into physics. Requiring a number in front leaves those as text.
# It also keeps "1206" out, which parses as a dimensionless 1206.
_NUMBER_THEN_UNIT = re.compile(r"^[-+]?(?:\d+\.?\d*|\.\d+)(?:[eE][-+]?\d+)?\s*\S")
_STRINGS = frozenset(
{
"case",
"temperature_coefficient_code",
"lifecycle",
}
)
def _coerce(column: str, text: str) -> Any:
"""Turn one CSV cell into an attribute value.
Dimensioned columns become quantities, so the unit travels with the number.
Booleans become real bools, since the vocabulary says bool and the category's
polarity check refuses a non-bool rather than consulting truthiness. A column
the vocabulary does not name still becomes a quantity when it carries its own
unit, so a provider's own dimensioned attribute is queryable like any other.
"""
text = text.strip()
if not text:
return None
if column in _BOOLS:
return text.lower() in ("1", "true", "yes", "y")
if column in _STRINGS:
return text
if column in _UNITS:
# Bare number first: "10000" parses as a *dimensionless* quantity, which
# is exactly the silent wrongness carrying units is supposed to prevent.
try:
magnitude = float(text)
except ValueError:
# The cell states its own unit, so use it as written. Somebody who
# wrote "0.1 uF" meant it to read that way, and rescaling it would be
# overruling them.
return registry.Quantity(text)
return _prefixed(magnitude, _UNITS[column])
# Past the vocabulary, a CSV carries no type information and numeric-looking
# text is a guess. Leading-zero text stays a string, which is a weak safety
# net for code-like columns nobody has named; a provider with a real schema
# would not be guessing at all.
if len(text) > 1 and text[0] == "0" and text[1].isdigit():
return text
try:
return int(text)
except ValueError:
pass
try:
return float(text)
except ValueError:
pass
# Not a plain number, so it may still be a dimensioned one. A provider's own
# column is exactly where a unit the core has never heard of shows up, and
# reading it as text would make it unqueryable by any dimensioned predicate.
if _NUMBER_THEN_UNIT.match(text):
try:
return registry.Quantity(text)
except Exception: # noqa: BLE001 - see below
# Broad on purpose. The try holds one call into a third-party parser
# fed arbitrary text from a file, and every failure means the same
# thing: not a quantity, so treat it as text. pint raises across at
# least PintError, TokenError and AssertionError depending on how the
# cell is malformed, and that set is neither documented nor stable --
# naming a subset would turn an unlisted one into a failed build.
pass
return text
def _sort_key(value: Any) -> tuple[int, float, str]:
"""Order values of unknown type: numbers first by magnitude, then strings."""
if isinstance(value, PlainQuantity):
return (0, float(value.to_base_units().magnitude), "")
if isinstance(value, bool):
return (0, float(value), "")
if isinstance(value, (int, float)):
return (0, float(value), "")
return (1, 0.0, str(value))
def _attribute_text(identity: PartIdentity, key: str) -> str:
"""Read a string attribute off the identity, or say it is not ours.
An entry comes back from a lockfile a person is invited to edit, so its shape
is an assumption rather than a fact, and `StalePayload` is what a provider
raises when it cannot honour what it is handed. An absent attribute reads as
empty, which is what a catalogue without that column always produced.
"""
value = identity.attributes.get(key, "")
if not isinstance(value, str):
raise StalePayload(
f"attribute {key!r} is {type(value).__name__}, expected a string"
)
return value
def _prefixed(magnitude: float, unit: str) -> Any:
"""Give a bare number an engineering prefix, so it reads as a person would.
A column of raw farads is convenient to maintain and terrible to look at: the
value ends up on the schematic, and `1e-07 F` is not what anyone calls that
part. `to_compact` picks the prefix; the rounding is because the rescale is a
float division and lands on `99.99999999999999 nanofarad` without it.
"""
compact = registry.Quantity(magnitude, unit).to_compact()
return registry.Quantity(round_significant(float(compact.magnitude)), compact.units)
@dataclass(frozen=True)
class _Row:
"""One CSV row: the core-defined identity, plus what this provider keeps."""
identity: PartIdentity
datasheet: str | None = None
[docs]
@dataclass(frozen=True)
class CsvPart:
"""A resolved part from a CSV row. Implements ResolvedPart."""
identity: PartIdentity
payload: dict[str, JSON]
[docs]
def materialize(self, component: Any) -> None:
"""Attach a landpattern and symbol from the standard generators."""
from .geometry import passive_geometry
# Geometry inputs come off the identity, which already carries them: the
# core stores `attributes` and `category` in the entry and hands them back
# to `rehydrate`, so a payload copy would be the same data twice.
identity = self.identity
case = _attribute_text(identity, "case")
polarized = bool(identity.attributes.get("polarized"))
landpattern, symbol = passive_geometry(identity.category, case, polarized)
component.landpattern = landpattern
component.symbol = symbol
# Everything the payload carries, not a named list of it: this channel is
# whatever the provider reports but cannot be queried on, and `datasheet`
# is just what this one happens to have. JITX declares no field for it, so
# it is a de facto attribute and a convention rather than a contract.
#
# Read back through `decode_payload`, which returns whatever was put in,
# quantities included. `_PRIVATE_PAYLOAD` stays off the component: it is
# bookkeeping for `rehydrate`, not something a design should see.
fields = decode_payload(self.payload)
apply_fields(
component,
{k: v for k, v in fields.items() if k not in _PRIVATE_PAYLOAD and v},
)
# Two-pin: JITX's default declaration-order mapping suffices, so no
# explicit PadMapping/SymbolMapping is needed. This branch is only ever
# reached with polarized=True from PolarizedCapacitor, whose ports are
# a/c: a plain Capacitor refuses a polarized part before materialize runs.
[docs]
@dataclass(frozen=True)
class CsvParts(CategoryProvider):
"""Resolve passives from a CSV of approved parts.
`category`, `mpn` and `manufacturer` are consumed as identity and `datasheet`
as payload; every other column becomes an attribute under its own name, so the
file's header is the vocabulary. Dimensioned columns are parsed into
quantities.
A row is a candidate when `match` says so, and candidates are ordered by the
query's `rank`. Filtering and ordering cover every column the file has, which
is the provider invariant: whatever it exposes, it serves.
"""
path: str
"""The file, as written. Relative paths are resolved against `root`."""
root: str | None = None
"""Directory `path` is relative to, when it came from project config.
Deliberately absent from `config()`, and so from the chain digest: where a
project happens to be checked out is not a design decision, and baking an
absolute path into the digest would give two developers different chains for
the same project and never let the lock match.
"""
package: str | None = None
"""Installed package `path` is a resource of, for a catalogue shipped by a
company-wide pip package. Read through `importlib.resources`, so it works from
a wheel, an editable install or a zip import alike.
Unlike `root`, this one *is* in `config()`: which package a catalogue comes
from is a design decision, where its files happen to land on disk is not.
"""
name: str = field(init=False, compare=False)
"""`csv:` and the file's basename, or the package and basename for a resource.
Two catalogues with the same basename in different directories collide, which
is legal; `PartSource.names()` suffixes duplicates positionally.
"""
def __post_init__(self):
# object.__setattr__ in a frozen dataclass' __post_init__ is the canonical
# exception to avoiding it.
if self.package is not None:
name = f"csv:{self.package}/{Path(self.path).name}"
else:
name = f"csv:{Path(self.path).name}"
object.__setattr__(self, "name", name)
[docs]
@classmethod
def from_package(cls, package: str, resource: str):
"""A catalogue shipped inside an installed package.
The spelling for a company-wide approved-parts list: publish a chain from
the package that carries the CSV, and name the file as a resource rather
than as a path.
PartSource([CsvParts.from_package("acme_parts", "approved.csv")])
The package name and resource reach the chain digest; the filesystem
location does not, so every developer and every CI box computes the same
chain for the same installed version.
"""
return cls(resource, package=package)
[docs]
def config(self) -> dict[str, JSON]:
"""What identifies this catalogue, and nothing that does not.
The path or resource name, and the package when there is one: changing
either is a decision. `root` is excluded, and so is wherever a package
resolves to on disk, so the digest is the same on every machine.
The package's *version* is excluded too, deliberately. A chain published
by a package is pinned by dependency version, not by the lockfile; asking
the lock to defend against an upgrade as well would pin the same thing
twice, in two files, with no way to tell which is authoritative.
"""
if self.package is not None:
return {"package": self.package, "resource": self.path}
return {"path": self.path}
@property
def file(self) -> Any:
"""Where to read from: a package resource, or a filesystem path.
A `Traversable` in the package case rather than a `Path`, since the file
need not be on the filesystem at all.
"""
if self.package is not None:
return resources.files(self.package) / self.path
target = Path(self.path)
if self.root is not None and not target.is_absolute():
return Path(self.root) / target
return target
@functools.cached_property
def _rows(self) -> tuple[_Row, ...]:
# read_text() rather than open(): it is the part of the Traversable
# protocol every resource loader implements, so this works for a file on
# disk and for a resource inside a wheel without branching.
try:
content = self.file.read_text(encoding="utf-8")
except (FileNotFoundError, NotADirectoryError, IsADirectoryError, OSError):
return ()
out: list[_Row] = []
for raw in csv.DictReader(io.StringIO(content)):
attributes = {}
for column, cell in raw.items():
if column is None or column in _RESERVED_COLUMNS:
continue
value = _coerce(column, cell or "")
if value is not None:
attributes[column] = value
out.append(
_Row(
identity=PartIdentity(
mpn=(raw.get("mpn") or "").strip(),
manufacturer=raw.get("manufacturer"),
category=(raw.get("category") or "").strip(),
provider=self.name,
provider_ref=(raw.get("mpn") or "").strip(),
attributes=attributes,
),
datasheet=(raw.get("datasheet") or "").strip() or None,
)
)
return tuple(out)
@functools.cached_property
def _columns(self) -> frozenset[str]:
return frozenset(
key for row in self._rows for key in row.identity.attributes
) | frozenset({"mpn", "manufacturer", "category"})
def _unserved(self, query: CanonicalQuery) -> tuple[str, ...]:
"""Query keys this file has no column for.
Declined per query rather than declared in advance: there is no manifest,
so this is a statement about what the code can do with the file it has.
"""
wanted = set(query.filters) | {r.key for r in query.rank}
return tuple(sorted(wanted - self._columns))
[docs]
def resolve_one(self, query: CanonicalQuery) -> Outcome:
unserved = self._unserved(query)
if unserved:
return Unsupported(unserved, f"{self.file} has no such column")
rows = [r for r in self._rows if r.identity.category == query.category]
if not rows:
return Unsupported((), f"{self.file} holds no {query.category} rows")
candidates = [
r for r in rows if match(query.filters, r.identity.attributes, r.identity)
]
if not candidates:
return NoMatch(
f"0 of {len(rows)} {query.category} rows matched {query.describe()}"
)
ordered = rank_by(candidates, query.rank, key=lambda r: r.identity.attributes)
return Resolved(tuple(self._part(r) for r in ordered[: max(1, query.limit)]))
[docs]
def distinct(self, query: CanonicalQuery, key: str) -> Sequence[object]:
"""Values of `key` among matching rows, in sorted order."""
if key not in self._columns:
return ()
seen: list[Any] = []
for row in self._rows:
identity = row.identity
if identity.category != query.category:
continue
if not match(query.filters, identity.attributes, identity):
continue
value = identity.attributes.get(key)
if value is not None and value not in seen:
seen.append(value)
# A column's values are whatever the file holds, so they may not be
# mutually orderable. Sort on a key that is always comparable, falling
# back to file order for anything that still refuses.
try:
return sorted(seen, key=_sort_key)
except TypeError:
return seen
def _part(self, row: _Row) -> CsvPart:
# Only what is not already in the entry. `case`, `polarized` and the
# category reach `rehydrate` through the identity, so the payload carries
# the version and the one field that is presentation rather than
# selection.
#
# Built through `encode_payload`, which renders values as JSON so a
# provider does not hand-convert them. Nothing here is dimensioned today;
# a payload holding a quantity would work the same way.
return CsvPart(
identity=row.identity,
payload=encode_payload({"v": 1, "datasheet": row.datasheet}),
)
[docs]
def rehydrate(
self, identity: PartIdentity, payload: dict[str, JSON]
) -> ResolvedPart:
"""Rebuild a part from a lockfile entry.
The payload carries its own version, so this provider can disown a payload
it no longer understands rather than the core guessing.
"""
if payload.get("v") != 1:
raise StalePayload(
f"{self.name}: unknown payload version {payload.get('v')!r}"
)
return CsvPart(identity=identity, payload=dict(payload))
