"""
Exploration
===========
Asking a chain what it has, from a script, a REPL or a notebook. `search` returns
candidates keyed by the chain member that offered them, `explain` says what each
provider made of a query, and `distinct` lists the values a key takes::
source = PartSource([CsvParts("parts/approved-parts.csv")])
found = search(ResistorQuery(resistance=10 * kohm), source=source)
for member, parts in found.items():
for part in parts:
print(member, part.identity.mpn, part.identity.attributes)
# Flat, still in chain order, where the grouping does not matter:
for part in itertools.chain.from_iterable(found.values()):
print(part.identity.mpn)
print(explain(CapacitorQuery(capacitance=22 * uF), source=source))
print(distinct(CapacitorQuery(case="0402"), "rated_voltage_dc", source=source))
`explain` is the quickest way to find out why nothing matched: it prints the chain
in order with each provider's own reason. `source` may be omitted inside a
project, where the configured chain is discovered; a notebook with no project
around it has to pass one.
**Browsing walks the chain; choosing does not.** `search` gathers candidates from
member after member until it has the `limit` asked for, where `resolve` takes the
first member that answers and stops. Both are right for what they do: precedence
decides which part is *selected*, but a request for twenty is not answered by the
three the first member happened to hold. Candidates keep chain order and are never
compared across members, since ranking two providers' parts against each other
needs an agreed vocabulary this design does not impose.
**Nothing here is cached.** A person asking twice has usually changed something in
between, and exploration binds nothing and writes nothing, so there is no second
run for an answer to stay consistent with -- the resolver's cache exists for that
reason and this one had inherited the shape without the reason. Hold the list in a
variable if a slow chain is being asked the same thing repeatedly.
Inside a design, note that `source=None` discovers the *project's* chain and never
the active context, so it may not be the chain the design resolves with. Pass
`source=PartSource.get()` to explore the one actually in force.
Not for design code, and three deliberate properties keep it that way. These
return `ResolvedPart` and `PartIdentity`, never components, so browsing two
hundred candidates costs two hundred rows and not two hundred landpatterns. They
take an explicit `source` rather than reading a context. And nothing here can feed
a component: design code resolves exactly one part per component, which is why
`PartQuery` has no `limit`, and choosing between candidates in design code stays
impossible by construction, because that decision belongs to a provider, where it
is reproducible and lockable.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Any
from .error import NoChainConfigured
from .provider import PartSource, describe
from .query import PartQuery
from .record import ResolvedPart
def _source_for(source: PartSource | None) -> PartSource:
"""The chain to explore with.
Never the context: a notebook has no `Design`, so no instantiation is active
and `PartSource.require()` would hand back an `Instantiable` placeholder rather
than a chain.
"""
if source is not None:
return source
from .policy import current_project, discovered_source
try:
return discovered_source(current_project())
except NoChainConfigured as exc:
raise NoChainConfigured(
"no project here, so there is no chain to discover; pass source="
) from exc
[docs]
def search(
query: PartQuery,
limit: int = 20,
source: PartSource | None = None,
) -> Mapping[str, list[ResolvedPart]]:
"""Up to `limit` candidates, keyed by the member that offered them.
Returns parts, never components. Gathered along the whole chain rather than
from the first member that has anything: a member offering three of the twenty
asked for does not end the question. Keyed by member, in chain order, because
which catalogue has a part is usually the thing being asked. See
`PartSource.candidates` for what the keys do and do not promise.
Nothing is cached. Two calls ask twice, which is what a person editing a
catalogue between them wants, and exploration has no answer to keep stable --
it binds nothing and writes nothing, so there is no second run to agree with.
Hold the list in a variable if a slow chain is being asked the same thing
repeatedly.
"""
chain = _source_for(source)
return chain.candidates(query.canonical(limit=limit), limit)
[docs]
def explain(
query: PartQuery,
source: PartSource | None = None,
) -> str:
"""Why a query failed, or what each provider said about it.
The chain is shown in order, with where it came from and each provider's own
reason. That last part is what removing the capability manifest bought: a
reader gets "no price data for this manufacturer" rather than "the runtime
skipped this provider because it does not declare price".
"""
chain = _source_for(source)
canonical = query.canonical(limit=1)
lines = [f"chain from {_provenance(source)}"]
for chain_outcome in chain.outcomes([canonical]):
for index, (name, outcome) in enumerate(chain_outcome.trail, start=1):
lines.append(f" {index} {name:24s} {describe(outcome)}")
if not chain_outcome.trail:
lines.append(" (the chain is empty)")
return "\n".join(lines)
[docs]
def distinct(
query: PartQuery,
key: str,
source: PartSource | None = None,
) -> Mapping[str, Sequence[Any]]:
"""Values of `key` among matching parts, per provider that implements it.
Keyed by provider rather than flattened into a union, since which provider
knows a value is usually what the caller is trying to find out. Dimensioned
attributes come back as quantities, because providers carry units.
"""
chain = _source_for(source)
return chain.distinct(query.canonical(limit=1), key)
def _provenance(source: PartSource | None) -> str:
"""Where the chain being used came from, for `explain`'s first line."""
if source is not None:
return f"explicit source ({', '.join(source.names()) or 'empty'})"
from .policy import current_project
inside = current_project() is not None
return "project config" if inside else "installed providers"
__all__ = ["distinct", "explain", "search"]
