from __future__ import annotations
import re
from functools import lru_cache
from typing import Any
from ._data import TUTORIALS_DIR, load_model_parameters, load_schema
from .discriminator import known_funcs
from .models import (
ModelCatalog,
ModelInfo,
ModelParameters,
ParameterInfo,
SemanticMatch,
Tutorial,
TutorialSearchResult,
)
GROUP_PREFIX = {
"scene_generation": "SceneGeneration",
"photon_collection": "PhotonCollection",
"charge_collection": "ChargeCollection",
"charge_generation": "ChargeGeneration",
"charge_transfer": "ChargeTransfer",
"charge_measurement": "ChargeMeasurement",
"readout_electronics": "ReadoutElectronics",
"phasing": "Phasing",
"data_processing": "DataProcessing",
}
PREFIXES = tuple(GROUP_PREFIX.values())
def _split_base(base: str) -> tuple[str | None, str | None]:
for prefix in PREFIXES:
if base.startswith(f"{prefix}_"):
return prefix, base[len(prefix) + 1 :]
return None, None
@lru_cache(maxsize=1)
def _manifest() -> list[tuple[str, str, str]]:
entries: list[tuple[str, str, str]] = []
for path in sorted(TUTORIALS_DIR.glob("*.md")):
prefix, model = _split_base(path.stem)
if prefix and model:
entries.append((prefix, model, path.name))
return entries
def _title_of(text: str, fallback: str) -> str:
for line in text.splitlines():
line = line.strip()
if line.startswith("#"):
return line.lstrip("#").strip()
return fallback
[docs]
def find_tutorial(group: str, model: str) -> Tutorial | None:
"""Fetch the tutorial for one model.
Parameters
----------
group : str
The pipeline stage the model belongs to, such as
``"charge_generation"``. The tutorial file prefix, such as
``"ChargeGeneration"``, works too.
model : str
The model's short name, such as ``"dark_current"``.
Returns
-------
Tutorial or None
The whole tutorial as Markdown, or None when the catalog has no
tutorial for that model.
"""
prefix = GROUP_PREFIX.get(group.lower(), group)
wanted = f"{prefix}_{model}".lower()
for pfx, mdl, name in _manifest():
if f"{pfx}_{mdl}".lower() == wanted:
path = TUTORIALS_DIR / name
body = path.read_text(encoding="utf-8")
return Tutorial(
group=group,
model=mdl,
title=_title_of(body, mdl),
body_markdown=body,
)
return None
[docs]
def search_tutorials(query: str, limit: int = 5, semantic: bool | None = None) -> list[Tutorial]:
"""Whole tutorials matching a query.
`semantic` picks the engine: True ranks with the vector index, False
matches on keywords, and None (the default) uses the index only when one
has already been built. Whichever is asked for, a missing index or a
missing extra degrades to keywords rather than failing; callers who need
to know which engine answered use semantic_search_with_fallback.
Parameters
----------
query : str
What to look for, in plain words.
limit : int, default 5
The most tutorials to return.
semantic : bool or None, default None
Which engine to use, as above.
Returns
-------
list of Tutorial
The best matches first; empty when nothing matches.
"""
if semantic is not False:
try:
from . import rag
if semantic or rag.index_exists():
matches = rag.semantic_search(query, limit=limit)
results: list[Tutorial] = []
seen: set[tuple[str, str]] = set()
for m in matches:
key = (m.group, m.model)
if key in seen:
continue
seen.add(key)
tut = find_tutorial(m.group, m.model)
if tut:
results.append(tut)
if results:
return results
except Exception: # noqa: BLE001 - any engine failure means "use keywords"
pass
return _keyword_search(query, limit)
def _keyword_search(query: str, limit: int = 5) -> list[Tutorial]:
terms = [t for t in re.split(r"\W+", query.lower()) if t]
if not terms:
return []
scored: list[tuple[int, str, str, str]] = []
for prefix, model, name in _manifest():
body = (TUTORIALS_DIR / name).read_text(encoding="utf-8")
haystack = f"{name} {body}".lower()
name_l = name.lower()
score = 0
for term in terms:
if term in name_l:
score += 5
score += haystack.count(term)
if score:
scored.append((score, prefix, model, name))
scored.sort(key=lambda s: s[0], reverse=True)
prefix_to_group = {v: k for k, v in GROUP_PREFIX.items()}
results: list[Tutorial] = []
for _, prefix, model, name in scored[:limit]:
body = (TUTORIALS_DIR / name).read_text(encoding="utf-8")
results.append(
Tutorial(
group=prefix_to_group.get(prefix, prefix),
model=model,
title=_title_of(body, model),
body_markdown=body,
)
)
return results
#: The order Pyxel runs its stages in (pyxel/pipelines/pipeline.py,
#: MODEL_GROUPS), which picks the stage for a model name two stages share
#: (simple_collection) or one that runs in any stage (load_detector).
RUN_ORDER = (
"scene_generation",
"photon_collection",
"phasing",
"charge_generation",
"charge_collection",
"charge_transfer",
"charge_measurement",
"signal_transfer",
"readout_electronics",
"data_processing",
)
_TYPE_NAMES = {
"number": "float",
"integer": "int",
"string": "str",
"boolean": "bool",
"array": "list",
"object": "dict",
"null": "null",
}
def _schema_types(spec: dict[str, Any]) -> list[str]:
listed = spec.get("type")
types: list[str] = []
if isinstance(listed, list):
types = list(listed)
elif isinstance(listed, str):
types = [listed]
for key in ("anyOf", "oneOf"):
for variant in spec.get(key) or []:
if isinstance(variant, dict):
types += [t for t in _schema_types(variant) if t not in types]
return types
def _type_name(spec: dict[str, Any]) -> str:
"""A schema type as Python spells it: ``float``, ``list[int] | null``."""
names: list[str] = []
for t in _schema_types(spec):
name = _TYPE_NAMES.get(t, t)
items = spec.get("items")
if t == "array" and isinstance(items, dict) and _schema_types(items):
name = f"list[{_type_name(items)}]"
if name not in names:
names.append(name)
if isinstance(spec.get("enum"), list):
choices = ", ".join(str(v) for v in spec["enum"] if v is not None)
names = [f"one of {choices}"] + (["null"] if "null" in names else [])
return " | ".join(names) or "any"
def _model_definitions(model: str) -> list[tuple[str, dict[str, Any]]]:
"""(stage, arguments schema) for every stage the bundled schema has `model` in."""
found: list[tuple[str, dict[str, Any]]] = []
for definition in (load_schema().get("definitions") or {}).values():
props = definition.get("properties") if isinstance(definition, dict) else None
if not isinstance(props, dict):
continue
func_spec = props.get("func")
func = func_spec.get("const") if isinstance(func_spec, dict) else None
if not isinstance(func, str) or not func.startswith("pyxel.models."):
continue
parts = func.split(".")
if parts[-1] != model or len(parts) not in (3, 4):
continue
held = props.get("arguments")
arguments: dict[str, Any] = held if isinstance(held, dict) else {}
if len(parts) == 4:
found.append((parts[2], arguments))
else:
# pyxel.models.load_detector names no stage: it runs in any, and
# the schema lists it under each one it accepts it in.
stages = {m.stage for m in list_models().models if m.func == func}
found += [(stage, arguments) for stage in stages]
order = {stage: i for i, stage in enumerate(RUN_ORDER)}
return sorted(found, key=lambda pair: order.get(pair[0], len(order)))
[docs]
def model_parameters(model: str, stage: str | None = None) -> ModelParameters | None:
"""Look up a model's arguments, as the bundled Pyxel schema describes them.
Every model the schema accepts is covered. Each argument carries its type,
whether Pyxel requires it, Pyxel's own default when it has one, and the
example value Pyxel's example configs use for it, where there is one. The
example is not a default: a step that leaves an argument out gets Pyxel's
default, which for an argument such as simple_conversion's
quantum_efficiency is the detector's own value.
Parameters
----------
model : str
The model's short name, such as ``"load_image"``.
stage : str, optional
The pipeline stage, for a name two stages share (``simple_collection``).
Without it, the first stage in Pyxel's run order is taken.
Returns
-------
ModelParameters or None
The stage the model belongs to and each of its arguments, or None when
the schema has no such model (in that stage).
"""
found = [(s, args) for s, args in _model_definitions(model) if stage in (None, s)]
if not found:
return None
owning_stage, arguments = found[0]
examples = load_model_parameters().get(owning_stage, {}).get(model, {})
required = set(arguments.get("required") or [])
infos: list[ParameterInfo] = []
for name, spec in (arguments.get("properties") or {}).items():
if not isinstance(spec, dict):
continue
example = examples.get(name, {}).get("default") if isinstance(examples, dict) else None
infos.append(
ParameterInfo(
name=name,
type=_type_name(spec),
required=name in required,
has_default="default" in spec,
default=spec.get("default"),
example=example,
)
)
return ModelParameters(stage=owning_stage, model=model, parameters=infos)
[docs]
def list_models(stage: str | None = None) -> ModelCatalog:
"""List the models the bundled schema accepts.
Parameters
----------
stage : str, optional
One pipeline stage, such as ``"charge_generation"``. Leave it out to
list every stage. A stage the schema does not have lists nothing.
Returns
-------
ModelCatalog
Each model with its stage and the full ``func`` path to write in a
config.
"""
known = known_funcs()
infos: list[ModelInfo] = []
if stage is not None:
for func in sorted(known.funcs_by_stage.get(stage, set())):
infos.append(ModelInfo(stage=stage, func=func))
else:
for stg in sorted(known.funcs_by_stage):
for func in sorted(known.funcs_by_stage[stg]):
infos.append(ModelInfo(stage=stg, func=func))
return ModelCatalog(stage=stage, count=len(infos), models=infos)
def _snippet(text: str, length: int = 400) -> str:
return " ".join(text.split())[:length]
def _tutorial_to_match(tutorial: Tutorial) -> SemanticMatch:
return SemanticMatch(
group=tutorial.group,
model=tutorial.model,
title=tutorial.title,
section=tutorial.title,
score=0.0,
snippet=_snippet(tutorial.body_markdown),
)
[docs]
def semantic_search_with_fallback(query: str, limit: int = 5) -> TutorialSearchResult:
"""Search tutorial passages by meaning, and say which engine answered.
The vector index answers when it has been built. When it has not, or the
``rag`` extra is missing or fails, keyword matching answers instead and
the result is marked as degraded, so a caller always gets matches and
never an exception.
Parameters
----------
query : str
What to look for, in plain words.
limit : int, default 5
The most passages to return.
Returns
-------
TutorialSearchResult
The matches, the engine that found them, and the reason whenever
keyword matching stood in.
"""
reason: str | None = None
try:
from . import rag
if rag.index_exists():
matches = rag.semantic_search(query, limit=limit)
return TutorialSearchResult(mode="semantic", degraded=False, matches=matches)
reason = "vector index not built"
except Exception as exc:
# The cause itself (an old qdrant-client, a missing extra), so the
# degraded result says what to fix, not only that something failed.
detail = (str(exc).strip().splitlines() or [""])[0]
reason = f"{type(exc).__name__}: {detail}" if detail else type(exc).__name__
fallback = [_tutorial_to_match(t) for t in _keyword_search(query, limit)]
return TutorialSearchResult(
mode="keyword",
degraded=True,
message=f"Semantic index unavailable ({reason}); showing keyword results instead.",
matches=fallback,
)
