from __future__ import annotations
import copy
from collections.abc import Iterable
from functools import lru_cache
from typing import Any
from ._data import load_schema
from .models import ConfigFix, Diagnostic, FixResult
class KnownFuncs:
__slots__ = ("funcs_by_stage", "stage_by_func", "all_funcs", "schema_by_func")
def __init__(self) -> None:
self.funcs_by_stage: dict[str, set[str]] = {}
self.stage_by_func: dict[str, str] = {}
self.all_funcs: set[str] = set()
#: The resolved model definition for each func, so a pipeline step can
#: be validated against the one branch its func names instead of the
#: whole anyOf.
self.schema_by_func: dict[str, dict[str, Any]] = {}
def add(
self, stage: str, func: str | None, model_schema: dict[str, Any] | None = None
) -> None:
if not func:
return
self.funcs_by_stage.setdefault(stage, set()).add(func)
self.stage_by_func.setdefault(func, stage)
self.all_funcs.add(func)
if model_schema is not None:
self.schema_by_func.setdefault(func, model_schema)
def _resolve_ref(ref: object, schema: dict[str, Any]) -> dict[str, Any] | None:
if not isinstance(ref, str) or not ref.startswith("#/"):
return None
node: object = schema
for part in ref[2:].split("/"):
if not isinstance(node, dict):
return None
node = node.get(part)
return node if isinstance(node, dict) else None
def _func_const_of(model_schema: dict[str, Any] | None) -> str | None:
if not isinstance(model_schema, dict):
return None
func = (model_schema.get("properties") or {}).get("func")
if not isinstance(func, dict):
return None
const = func.get("const")
if isinstance(const, str):
return const
enum = func.get("enum")
if isinstance(enum, list) and len(enum) == 1 and isinstance(enum[0], str):
return str(enum[0])
return None
def _branches_of(item_schema: dict[str, Any] | None) -> list[dict[str, Any]]:
if not isinstance(item_schema, dict):
return []
for combiner in ("anyOf", "oneOf"):
branches = item_schema.get(combiner)
if isinstance(branches, list):
return branches
return [item_schema]
@lru_cache(maxsize=1)
def known_funcs() -> KnownFuncs:
schema = load_schema()
known = KnownFuncs()
for definition in (schema.get("definitions") or {}).values():
props = definition.get("properties") if isinstance(definition, dict) else None
if not isinstance(props, dict):
continue
for stage, prop_schema in props.items():
items = prop_schema.get("items") if isinstance(prop_schema, dict) else None
if not items:
continue
for branch in _branches_of(items):
model = _resolve_ref(branch.get("$ref"), schema) or branch
known.add(stage, _func_const_of(model), model)
return known
def _leaf(func: object) -> str:
return str(func).split(".")[-1]
def levenshtein(a: str = "", b: str = "") -> int:
a, b = str(a), str(b)
m, n = len(a), len(b)
if m == 0:
return n
if n == 0:
return m
prev = list(range(n + 1))
curr = [0] * (n + 1)
for i in range(1, m + 1):
curr[0] = i
for j in range(1, n + 1):
cost = 0 if a[i - 1] == b[j - 1] else 1
curr[j] = min(prev[j] + 1, curr[j - 1] + 1, prev[j - 1] + cost)
prev, curr = curr, prev
return prev[n]
[docs]
def suggest_func(bad_func: str, candidates: Iterable[str]) -> str | None:
"""Suggest the valid model a mistyped ``func`` most likely meant.
Names are compared by edit distance, both on the last part of the dotted
path and on the whole path. A candidate is only suggested when it is
within two edits, or two in five characters of that last part for a
longer name, so a name that is nothing like any model gets no suggestion.
Parameters
----------
bad_func : str
The ``func`` value as written in the config.
candidates : iterable of str
The valid ``func`` paths to choose from.
Returns
-------
str or None
The closest candidate, or None when none is close enough.
"""
options = list(candidates or [])
if not options:
return None
bad_leaf = _leaf(bad_func)
best: str | None = None
best_score = float("inf")
for cand in options:
score = min(levenshtein(bad_leaf, _leaf(cand)), levenshtein(bad_func, cand))
if score < best_score:
best_score, best = score, cand
tolerance = max(2, -(-len(bad_leaf) * 2 // 5)) # ceil(len * 0.4)
return best if best_score <= tolerance else None
class Finding:
__slots__ = (
"stage",
"index",
"func",
"path",
"reason",
"suggestion",
"owning_stage",
"valid_funcs",
)
def __init__(
self,
stage: str,
index: int,
func: str,
reason: str,
suggestion: str | None,
owning_stage: str | None,
valid_funcs: list[str],
) -> None:
self.stage = stage
self.index = index
self.func = func
self.path = f"/pipeline/{stage}/{index}/func"
self.reason = reason
self.suggestion = suggestion
self.owning_stage = owning_stage
self.valid_funcs = valid_funcs
@property
def base(self) -> str:
return f"/pipeline/{self.stage}/{self.index}"
#: Every model Pyxel ships lives under this namespace.
PYXEL_NAMESPACE = "pyxel."
def is_user_model(func: str) -> bool:
"""A model of the user's own, outside Pyxel.
Pyxel's schema accepts any func outside ``pyxel.models.`` through its
generic ModelFunction branch, and Pyxel imports it at run time, so it is
neither unknown nor something to rename into a built-in. A func that does
sit under ``pyxel.`` but is not a real model is still reported: Pyxel has
no models anywhere else, so it is a typo.
"""
return "." in func and not func.startswith(PYXEL_NAMESPACE)
[docs]
def find_unknown_funcs(config: object, known: KnownFuncs | None = None) -> list[Finding]:
"""Find the pipeline steps whose ``func`` is not a model of their stage.
A model of the user's own, outside the ``pyxel.`` namespace, is left
alone: Pyxel imports it at run time.
Parameters
----------
config : object
A parsed config. Anything without a ``pipeline`` mapping has no
findings.
known : KnownFuncs, optional
The models to check against. By default, the ones the bundled schema
defines.
Returns
-------
list of Finding
One per step, each with its ``reason`` (``"unknown"`` for a name no
stage has, ``"wrong-stage"`` for a model placed in another stage), the
``owning_stage`` of a misplaced model, a ``suggestion`` for a likely
typo, and the ``valid_funcs`` of the step's stage.
"""
if not isinstance(config, dict):
return []
pipeline = config.get("pipeline")
if not isinstance(pipeline, dict):
return []
known = known or known_funcs()
findings: list[Finding] = []
for stage, steps in pipeline.items():
if not isinstance(steps, list):
continue
valid = known.funcs_by_stage.get(stage, set())
for index, step in enumerate(steps):
if not isinstance(step, dict):
continue
func = step.get("func")
if not isinstance(func, str):
continue
if func in valid:
continue
if is_user_model(func):
continue
owning_stage = known.stage_by_func.get(func)
reason = "wrong-stage" if owning_stage else "unknown"
candidates = valid or known.all_funcs
findings.append(
Finding(
stage=stage,
index=index,
func=func,
reason=reason,
suggestion=None if reason == "wrong-stage" else suggest_func(func, candidates),
owning_stage=owning_stage,
valid_funcs=sorted(valid),
)
)
return findings
def _describe(finding: Finding) -> str:
if finding.reason == "wrong-stage":
return (
f"Model '{_leaf(finding.func)}' is valid, but not in stage "
f"'{finding.stage}'. It belongs to '{finding.owning_stage}'."
)
msg = f"Unknown model '{_leaf(finding.func)}' in stage '{finding.stage}'."
if finding.suggestion:
msg += f" Did you mean '{_leaf(finding.suggestion)}'?"
return msg
def _hint(finding: Finding) -> str:
if finding.reason == "wrong-stage":
return f"Move this step under pipeline -> {finding.owning_stage}."
return "Check the function name against the valid models for this stage."
def next_step_for(finding: Finding) -> str:
"""The one instruction explain_config attaches to a func finding."""
if finding.reason == "wrong-stage" and finding.owning_stage:
return f"Move this step to pipeline.{finding.owning_stage}."
if finding.suggestion:
return f"Rename func to '{finding.suggestion}'."
if finding.valid_funcs:
return f"Pick one of the {len(finding.valid_funcs)} models valid in '{finding.stage}'."
return f"'{finding.stage}' is not a pipeline stage; move its steps under a real stage."
def _diagnostic(finding: Finding, *, with_next_step: bool) -> Diagnostic:
shown = [_leaf(f) for f in finding.valid_funcs[:8]]
return Diagnostic(
severity="error",
path=finding.path,
message=_describe(finding),
hint=_hint(finding),
suggestion=_leaf(finding.suggestion) if finding.suggestion else None,
valid_options=shown or None,
next_step=next_step_for(finding) if with_next_step else None,
category="func",
)
def diagnostics_for(
config: object, known: KnownFuncs | None = None, *, with_next_step: bool = False
) -> list[Diagnostic]:
return [
_diagnostic(f, with_next_step=with_next_step) for f in find_unknown_funcs(config, known)
]
def offending_bases(config: object, known: KnownFuncs | None = None) -> set[str]:
return {f.base for f in find_unknown_funcs(config, known)}
[docs]
def autofix(config: dict[str, Any] | str) -> FixResult:
"""Repair mistyped and misplaced model names in a config.
A mistyped ``func`` is renamed to its closest valid model, and a valid
model in the wrong stage is moved to the stage it belongs to. Only
``func`` names change: required arguments are never invented, so any
that are still missing appear in ``remaining_diagnostics``. The config
that is passed in is never modified.
Parameters
----------
config : dict or str
The config as a mapping, or the raw YAML text of a ``.yaml`` file.
Returns
-------
FixResult
The corrected config, each change made, what could not be repaired,
and the full validation findings for the corrected config. Text that
cannot be read as a config comes back unresolved, with an empty
config.
"""
if isinstance(config, str):
from .validation import _as_config
from .yaml_read import ConfigReadError, read_problem
try:
config = _as_config(config)
except ConfigReadError as exc:
problem = read_problem(exc)
return FixResult(
changed=False,
changes=[],
unresolved=[problem],
remaining_diagnostics=[problem],
config={},
)
fixed = copy.deepcopy(config)
known = known_funcs()
changes: list[ConfigFix] = []
unresolved: list[Diagnostic] = []
for pipeline, prefix, label in _pipelines(fixed):
found_changes, found_unresolved = _fix_pipeline(pipeline, known)
changes += [
change.model_copy(update={"path": prefix + change.path}) for change in found_changes
]
unresolved += [
d.model_copy(update={"path": prefix + d.path, "message": label + d.message})
for d in found_unresolved
]
from .validation import validate_config
# Always the full verdict, so a config with nothing this can repair still
# reports what is wrong with it.
remaining = validate_config(fixed).diagnostics
return FixResult(
changed=bool(changes),
changes=changes,
unresolved=unresolved,
remaining_diagnostics=remaining,
config=fixed,
)
def _pipelines(config: dict[str, Any]) -> list[tuple[dict[str, Any], str, str]]:
"""(pipeline, path prefix, message prefix) for every pipeline in a config.
A flat config has one, at the root. A focal plane has one per tile, at
focal_plane_array/detectors/<id>/pipeline.
"""
if "focal_plane_array" in config:
plane = config["focal_plane_array"]
detectors = plane.get("detectors") if isinstance(plane, dict) else None
if not isinstance(detectors, dict):
return []
return [
(tile["pipeline"], f"/focal_plane_array/detectors/{tile_id}", f"Tile {tile_id}: ")
for tile_id, tile in detectors.items()
if isinstance(tile, dict) and isinstance(tile.get("pipeline"), dict)
]
pipeline = config.get("pipeline")
return [(pipeline, "", "")] if isinstance(pipeline, dict) else []
def _fix_pipeline(
pipeline: dict[str, Any], known: KnownFuncs
) -> tuple[list[ConfigFix], list[Diagnostic]]:
"""Rename and move the steps of one pipeline in place, with paths from its root."""
findings = find_unknown_funcs({"pipeline": pipeline}, known)
changes: list[ConfigFix] = []
unresolved: list[Diagnostic] = []
pending_moves: list[tuple[str, int, str]] = []
for finding in findings:
if finding.reason == "unknown" and finding.suggestion:
pipeline[finding.stage][finding.index]["func"] = finding.suggestion
changes.append(
ConfigFix(
path=finding.path,
action="rename_func",
before=finding.func,
after=finding.suggestion,
detail=(
f"Renamed unknown model to nearest valid match in stage "
f"'{finding.stage}'."
),
)
)
elif finding.reason == "wrong-stage" and finding.owning_stage:
pending_moves.append((finding.stage, finding.index, finding.owning_stage))
changes.append(
ConfigFix(
path=finding.path,
action="move_stage",
before=finding.stage,
after=finding.owning_stage,
detail=(
f"Moved model '{_leaf(finding.func)}' to its owning stage "
f"'{finding.owning_stage}'."
),
)
)
else:
unresolved.append(_diagnostic(finding, with_next_step=False))
if pending_moves:
removals: dict[str, list[int]] = {}
additions: dict[str, list[Any]] = {}
for source, index, destination in pending_moves:
removals.setdefault(source, []).append(index)
additions.setdefault(destination, []).append(pipeline[source][index])
for source, indices in removals.items():
for index in sorted(indices, reverse=True):
pipeline[source].pop(index)
# A misspelled stage key ("charge_generaton") is how every step
# under it came to be "in the wrong stage". Once emptied it is only
# an unknown property the schema will still reject, so it goes too.
# A real stage left empty is valid and is kept.
if not pipeline[source] and source not in known.funcs_by_stage:
del pipeline[source]
for destination, steps in additions.items():
if not isinstance(pipeline.get(destination), list):
pipeline[destination] = []
pipeline[destination].extend(steps)
return changes, unresolved
