from __future__ import annotations
import copy
import re
import sys
import tempfile
import warnings
from collections.abc import Callable, Iterable, Iterator
from pathlib import Path
from typing import Any
import yaml
from jsonschema import Draft7Validator
from jsonschema.exceptions import ValidationError
from ._data import load_schema
from .config_shape import (
ShapeDiagnosis,
closest_key,
diagnose_shape,
root_branch_name,
shape_diagnostics,
)
from .diagnostics import soft_warnings
from .discriminator import diagnostics_for as func_diagnostics
from .discriminator import find_unknown_funcs, known_funcs, next_step_for
from .expressions import Expression, evaluated_text
from .fpa import FPA_KEY, is_focal_plane, plane_diagnostics, tiles
from .models import Diagnostic, Explanation, ValidationResult
from .run_rules import run_rule_diagnostics, text_number_diagnostics
from .yaml_dialect import yaml11_spelling
from .yaml_read import ConfigReadError, read_config, read_problem
_DETECTOR_KEYS = ("ccd_detector", "cmos_detector", "mkid_detector", "apd_detector")
_MODE_KEYS = ("exposure", "observation", "calibration")
_SELECTOR_KEYS = _DETECTOR_KEYS + _MODE_KEYS
#: How many distinct paths a report lists before it summarises the rest. Enough
#: to see the shape of what is wrong; few enough that the first screen is not
#: the same field's anyOf alternatives repeated.
MAX_PATHS = 8
#: Fields Pyxel replaced between releases: (new name, release, the key the old
#: value goes under in the new field's mapping, or None when it moves as is).
#: A config written for the old name is the most common migration error, and
#: the schema can only say "unexpected". charge_to_volt_conversion was a bare
#: factor; charge_to_volt is a mapping, so a plain rename stays invalid.
_RENAMED: dict[str, tuple[str, str, str | None]] = {
"charge_to_volt_conversion": ("charge_to_volt", "Pyxel 3.0", "value"),
}
#: jsonschema phrases a type mismatch as "X is not of type 'a'" for one type
#: and "X is not of type 'a', 'b'" for a list; an anyOf of types produces one
#: message per branch. All three end up as one line naming every accepted type.
_TYPE_MESSAGE = re.compile(
r"^(?P<value>.*) is not of type (?P<types>'[^']+'(?:, '[^']+')*)$", re.DOTALL
)
def _as_config(config: dict[str, Any] | str) -> dict[str, Any]:
"""The config as a mapping; text is read by yaml_read.read_config."""
if isinstance(config, dict):
return config
return read_config(config)
def _path_of(error: ValidationError, prefix: str = "") -> str:
return prefix + "/" + "/".join(str(p) for p in error.absolute_path)
def _leaves(error: ValidationError) -> Iterator[ValidationError]:
if error.context:
for sub in error.context:
yield from _leaves(sub)
else:
yield error
def _is_branch_noise(error: ValidationError) -> bool:
path = list(error.absolute_path)
if path and path[-1] == "func":
return True
if path:
return False
msg = error.message
if "is a required property" in msg and any(k in msg for k in _SELECTOR_KEYS):
return True
if "Additional properties are not allowed" in msg and any(k in msg for k in _SELECTOR_KEYS):
return True
return False
def _is_root_structure(error: ValidationError) -> bool:
"""A root-level required/additionalProperties error.
Whatever it says, the config-shape check has already said it better: a
missing detector, two running modes, or a misspelled top-level key each
surface here as one line per schema branch.
"""
return not list(error.absolute_path) and error.validator in (
"required",
"additionalProperties",
)
def _under_base(path: str, bases: set[str]) -> bool:
return any(path == base or path.startswith(base + "/") for base in bases)
class _Leaf:
"""A leaf error with its path already prefixed, so re-validated pipeline
steps and root-level errors sort and group as one list."""
__slots__ = ("path", "error")
def __init__(self, path: str, error: ValidationError) -> None:
self.path = path
self.error = error
@property
def depth(self) -> int:
return self.path.count("/") if self.path != "/" else 0
def _known_step_bases(data: dict[str, Any]) -> dict[str, dict[str, Any]]:
"""Base path -> model definition, for every step whose func the schema knows."""
known = known_funcs()
pipeline = data.get("pipeline")
if not isinstance(pipeline, dict):
return {}
out: dict[str, dict[str, Any]] = {}
for stage, steps in pipeline.items():
if not isinstance(steps, list):
continue
for index, step in enumerate(steps):
if not isinstance(step, dict):
continue
func = step.get("func")
if isinstance(func, str) and func in known.funcs_by_stage.get(stage, set()):
out[f"/pipeline/{stage}/{index}"] = known.schema_by_func[func]
return out
def _step_at(data: dict[str, Any], base: str) -> Any:
_, _, stage, index = base.split("/")
return data["pipeline"][stage][int(index)]
def _raw_leaves(data: dict[str, Any], diagnosis: ShapeDiagnosis) -> list[_Leaf]:
"""Every leaf schema error, after the two anyOf explosions are defused.
The root anyOf is resolved by validating against the one branch the
chosen mode and detector name. The per-step anyOf is resolved by
re-validating each step whose func is known against that func's own
definition, so a mistyped argument reports as itself and not as the
missing required arguments of every other model in the stage.
"""
validator = Draft7Validator(load_schema())
branch = root_branch_name(diagnosis)
root = validator.evolve(schema=load_schema()["definitions"][branch]) if branch else validator
leaves = [
_Leaf(_path_of(leaf), leaf)
for error in root.iter_errors(data)
for leaf in _leaves(error)
if not _is_root_structure(leaf)
]
step_bases = _known_step_bases(data)
touched = {
base for base in step_bases if any(_under_base(leaf.path, {base}) for leaf in leaves)
}
if touched:
leaves = [leaf for leaf in leaves if not _under_base(leaf.path, touched)]
for base in sorted(touched):
step_validator = validator.evolve(schema=step_bases[base])
for error in step_validator.iter_errors(_step_at(data, base)):
for leaf in _leaves(error):
leaves.append(_Leaf(_path_of(leaf, prefix=base), leaf))
return leaves
def _unexpected_keys(error: ValidationError) -> list[str]:
if error.validator != "additionalProperties" or not isinstance(error.instance, dict):
return []
schema = error.schema if isinstance(error.schema, dict) else {}
allowed = set((schema.get("properties") or {}).keys())
return [str(k) for k in error.instance if str(k) not in allowed]
def _allowed_keys(error: ValidationError) -> list[str]:
schema = error.schema if isinstance(error.schema, dict) else {}
return sorted((schema.get("properties") or {}).keys())
def _replacement_step(key: str, value: object) -> str:
"""How to write an old field's value under the field that replaced it."""
new, release, inner = _RENAMED[key]
if inner is None:
return f"{release} renamed {key} to {new}; rename it."
if isinstance(value, (int, float)) and not isinstance(value, bool):
# Spelled so PyYAML, and so Pyxel, reads it back as a number (1.0e-06).
written = yaml11_spelling(repr(value)) if "e" in repr(value) else repr(value)
return (
f"{release} replaced {key} with {new}, which takes a mapping: write "
f"{new}: {{{inner}: {written}}}."
)
return (
f"{release} replaced {key} with {new}, which takes a mapping: put the value "
f"under {new} -> {inner}."
)
def _describe_unexpected(error: ValidationError) -> tuple[str | None, str | None]:
"""(suggestion, next_step) for an additionalProperties error."""
keys = _unexpected_keys(error)
if not keys:
return None, None
for key in keys:
if key in _RENAMED:
return _RENAMED[key][0], _replacement_step(key, error.instance.get(key))
allowed = _allowed_keys(error)
for key in keys:
guess = closest_key(key, allowed)
if guess:
return guess, f"Rename '{key}' to '{guess}'."
listed = ", ".join(f"'{k}'" for k in keys)
return None, f"Remove {listed}; the schema does not allow it here."
def _type_alternatives(messages: list[str]) -> tuple[str, list[str], list[str]]:
"""Split one path's messages into (the value, accepted types, other messages)."""
types: list[str] = []
others: list[str] = []
value = ""
for message in messages:
found = _TYPE_MESSAGE.match(message)
if found and (not types or found.group("value") == value):
value = found.group("value")
for name in found.group("types").split(", "):
if name.strip("'") not in types:
types.append(name.strip("'"))
else:
others.append(message)
return value, types, others
def _collapse(leaves: list[_Leaf], suppress_bases: set[str] | None = None) -> list[Diagnostic]:
bases = suppress_bases or set()
if bases:
leaves = [leaf for leaf in leaves if not _under_base(leaf.path, bases)]
signal = [leaf for leaf in leaves if not _is_branch_noise(leaf.error)]
if signal:
pool = signal
elif bases:
return []
else:
pool = leaves
# Group by path, keeping the first error seen for each distinct message so
# the structured fields (validator, instance, schema) are still at hand.
by_path: dict[str, list[ValidationError]] = {}
for leaf in pool:
errors = by_path.setdefault(leaf.path, [])
if all(leaf.error.message != e.message for e in errors):
errors.append(leaf.error)
# Deepest first: the field itself before the container it sits in.
ordered = sorted(by_path, key=lambda p: p.count("/") if p != "/" else 0, reverse=True)
out: list[Diagnostic] = []
for path in ordered[:MAX_PATHS]:
errors = by_path[path]
shown = "" if path == "/" else path
value, types, others = _type_alternatives([e.message for e in errors])
if types:
if len(types) == 1:
message = f"{value} is not of type '{types[0]}'"
else:
message = f"{value} is not one of the accepted types: {', '.join(types)}"
out.append(
Diagnostic(
severity="error",
path=shown,
message=message,
hint="Schema mismatch: fix this field.",
valid_options=types,
next_step=f"Give this field a value of type {' or '.join(types)}.",
category="schema",
)
)
for error in errors:
if error.message not in others:
continue
suggestion, next_step = _describe_unexpected(error)
out.append(
Diagnostic(
severity="error",
path=shown,
message=error.message,
hint="Schema mismatch: fix this field.",
suggestion=suggestion,
next_step=next_step,
category="schema",
)
)
hidden = ordered[MAX_PATHS:]
if hidden:
listed = ", ".join(hidden)
plural = "s" if len(hidden) != 1 else ""
out.append(
Diagnostic(
severity="error",
path="",
message=f"and {len(hidden)} more schema issue{plural} at: {listed}",
hint=f"Only the first {MAX_PATHS} paths are listed; fix those and re-run.",
category="schema",
)
)
return out
def _schema_diagnostics(data: dict[str, Any]) -> list[Diagnostic]:
diagnosis = diagnose_shape(data)
diagnostics: list[Diagnostic] = list(shape_diagnostics(data))
findings = find_unknown_funcs(data)
diagnostics.extend(func_diagnostics(data, with_next_step=True))
suppress = {f.base for f in findings}
leaves = _raw_leaves(data, diagnosis)
if leaves:
diagnostics.extend(_collapse(leaves, suppress_bases=suppress))
covered = {problem.label for problem in diagnosis.problems if problem.label}
diagnostics.extend(soft_warnings(data, already_reported=covered))
return diagnostics
def _moved_to_tile(d: Diagnostic, tile_id: Any, detector_key: str) -> Diagnostic | None:
"""A finding on a tile's flat config, at its place in the focal plane.
Pyxel builds each tile from the plane's running mode, the tile's detector
block and the tile's pipeline. A finding in the running mode belongs to
the plane and is said once; one in the detector or the pipeline names
its tile. Anything about the flat config's own root (no detector, two
running modes) is the plane check's to say, so it is dropped here.
"""
first = next((part for part in d.path.split("/") if part), "")
if first in _MODE_KEYS:
return d.model_copy(update={"path": f"/{FPA_KEY}{d.path}"})
tile = f"/{FPA_KEY}/detectors/{tile_id}"
if first in (detector_key, "pipeline"):
return d.model_copy(
update={"path": f"{tile}{d.path}", "message": f"Tile {tile_id}: {d.message}"}
)
if not d.path and d.category == "schema" and " more schema issue" in d.message:
# The overflow note lists the paths it did not show; they move too.
head, _, listed = d.message.partition(" at: ")
paths = ", ".join(f"{tile}{path}" for path in listed.split(", "))
return d.model_copy(update={"path": tile, "message": f"Tile {tile_id}: {head} at: {paths}"})
return None
def _on_chain(paths: list[str], path: str) -> int:
"""How many of `paths` are `path` or hold it (the document root aside)."""
return sum(1 for p in paths if p not in ("", "/") and (path == p or path.startswith(p + "/")))
def _put(config: Any, path: str, value: Any) -> None:
"""Replace the value at the JSON-pointer `path` in `config` itself."""
parts = [p.replace("~1", "/").replace("~0", "~") for p in path.split("/")[1:]]
node = config
for i, part in enumerate(parts):
last = i == len(parts) - 1
if isinstance(node, list):
index = int(part)
if last:
node[index] = value
node = node[index]
continue
# Mapping keys can be numbers (a focal plane's tile IDs).
key = next((k for k in node if str(k) == part), part)
if last:
node[key] = value
node = node[key]
def _text_numbers_refused(
config: dict[str, Any],
schema_errors: Callable[[Any], list[str]],
covered: Callable[[str], bool] = lambda _: True,
) -> list[Diagnostic]:
"""Numbers Pyxel reads as text, where that text is a problem.
1e3 is text to PyYAML, which is right for a field that takes only text (a
`name:`, a folder) and wrong anywhere a number belongs, including a field
that takes a number or a file name (charge_to_volt's value), where Pyxel
then looks for a file called 1e-6. So the schema decides: a value is left
alone only when the schema refuses it as the number it would be, more
than as the text it is written as (said at the value or at a union
holding it). Where no schema reaches (`covered` says which paths it does),
every such value is reported. `schema_errors` gives the paths of the
schema's errors for a config.
"""
candidates = text_number_diagnostics(config)
checked = [d for d in candidates if covered(d.path)]
if not checked:
return candidates
# One check with every such value as its number, compared path by path:
# a config can hold thousands of them (a long readout times list).
as_numbers = copy.deepcopy(config)
for d in checked:
_put(as_numbers, d.path, float(yaml11_spelling(str(_value_at(config, d.path)))))
base, after = schema_errors(config), schema_errors(as_numbers)
return [
d
for d in candidates
if not covered(d.path) or _on_chain(after, d.path) <= _on_chain(base, d.path)
]
def _value_at(config: Any, path: str) -> Any:
node = config
for part in [p.replace("~1", "/").replace("~0", "~") for p in path.split("/")[1:]]:
if isinstance(node, list):
node = node[int(part)]
else:
node = node[next((k for k in node if str(k) == part), part)]
return node
def _flat_schema_errors(config: Any) -> list[str]:
return [d.path for d in _schema_diagnostics(config) if d.severity == "error"]
def _plane_schema_errors(data: Any) -> list[str]:
"""The schema's error paths for every tile of a plane, moved into it."""
out: list[str] = []
for tile_id, detector_key, flat in tiles(data):
for d in _schema_diagnostics(flat):
moved = _moved_to_tile(d, tile_id, detector_key)
if moved is not None and moved.severity == "error":
out.append(moved.path)
return out
def _focal_plane_diagnostics(data: dict[str, Any]) -> list[Diagnostic]:
"""A focal plane: the plane itself, then every tile as Pyxel builds it.
The schema has no focal_plane_array branch (fpa.py), but each tile is the
flat config of the plane's running mode, its detector and its pipeline,
so it is validated as one, run rules included, and every finding is moved
back to where it sits in the plane. Numbers Pyxel reads as text are found
once over the whole document, the plane-wide scene included.
"""
out = plane_diagnostics(data)
seen = {(d.path, d.message) for d in out}
for tile_id, detector_key, flat in tiles(data):
found = _schema_diagnostics(flat) + run_rule_diagnostics(flat, include_text_numbers=False)
for d in found:
moved = _moved_to_tile(d, tile_id, detector_key)
if moved is None or (moved.path, moved.message) in seen:
continue
seen.add((moved.path, moved.message))
out.append(moved)
# The tiles and the plane's running mode are what the schema covers.
schema_covered = tuple(
f"/{FPA_KEY}/{part}/" for part in ("detectors", "exposure", "observation", "calibration")
)
out += [
d
for d in _text_numbers_refused(
data, _plane_schema_errors, lambda path: path.startswith(schema_covered)
)
if (d.path, d.message) not in seen
]
return out
#: Warnings Python hides from a user by default, because they are meant for
#: developers: Pyxel 3.0.2 raises "This class will be removed." from its own
#: pixel containers on every load, which says nothing about the config.
_DEVELOPER_WARNINGS = (
DeprecationWarning,
PendingDeprecationWarning,
ImportWarning,
ResourceWarning,
)
def _load_warnings(caught: list[warnings.WarningMessage]) -> list[Diagnostic]:
"""The warnings pyxel.load() raised that Python would show its user."""
out: list[Diagnostic] = []
seen: set[tuple[str, str]] = set()
for warning in caught:
if issubclass(warning.category, _DEVELOPER_WARNINGS):
continue
key = (warning.category.__name__, str(warning.message))
if key in seen:
continue
seen.add(key)
out.append(
Diagnostic(
severity="warning",
path="",
message=f"Pyxel warned while loading the config: {warning.message}",
hint=f"A {warning.category.__name__} from pyxel.load().",
category="pyxel-load",
)
)
return out
def _import_failed(exc: BaseException) -> Diagnostic:
first_line = (str(exc).strip().splitlines() or [""])[0]
return Diagnostic(
severity="warning",
path="",
message=(
"Deep check could not run: importing pyxel failed "
f"({type(exc).__name__}: {first_line})."
),
hint=(
"Pyxel is installed but could not be imported here; fix that environment, "
"or run the check where Pyxel imports."
),
category="pyxel-load",
)
def _refused(expression: Expression) -> Diagnostic:
text = expression.text
shown = text if len(text) <= 80 else text[:77] + "..."
if expression.kind == "binary":
what = f"this value is binary data, {shown}"
elif expression.kind == "file":
what = f"pyxel.load() would open {shown!r} while it loads the config"
else:
what = f"pyxel.load() would evaluate {shown!r} as Python"
return Diagnostic(
severity="warning",
path=expression.path,
message=f"Deep check refused to run: {what}, and {expression.reason}.",
hint=(
"The deep check runs the config, so it only runs what it knows is harmless. "
"For a config you trust, pass --trust-config (or trusted=True)."
),
category="pyxel-load",
)
[docs]
def deep_check(
config: dict[str, Any] | str, trusted: bool = False
) -> tuple[list[Diagnostic], bool]:
"""Run the config through pyxel.load() itself.
The schema is a description of Pyxel; Pyxel is the thing. Bounds it only
enforces in code (readout times out of order pass the schema and fail here)
are caught by handing the YAML to its own loader. Returns (diagnostics,
whether the loader actually ran).
Loading runs part of the config: Pyxel evaluates some text as Python,
unpickles binary data, imports the module each model names and opens a few
files (see expressions.py). Unless the caller says the config is trusted,
every such value is read first and the loader is not run if any of them is
more than plain data or a known harmless call.
Parameters
----------
config : dict or str
The config as a mapping, or the raw YAML text of a ``.yaml`` file.
trusted : bool, default False
Load the config without reading those values first. Only for a config
you wrote yourself.
Returns
-------
tuple of (list of Diagnostic, bool)
What the loader reported, and whether it actually ran. It does not run
when Pyxel is missing or cannot be imported, or when a value is
refused; the findings then say why.
"""
if not trusted:
# Text that cannot be read as a config is not handed to the loader.
try:
data = _as_config(config)
except ConfigReadError as exc:
return [read_problem(exc)], False
refused = [_refused(e) for e in evaluated_text(data)]
if refused:
return refused, False
try:
import pyxel
except ModuleNotFoundError as exc:
if exc.name != "pyxel":
return ([_import_failed(exc)], False)
return (
[
Diagnostic(
severity="warning",
path="",
message="Deep check skipped: pyxel is not installed.",
hint='Install it with pip install "pyxel-sim" to run the real loader.',
category="pyxel-load",
)
],
False,
)
except Exception as exc: # noqa: BLE001 - any failure to import is reported, never hidden
# Pyxel is installed but cannot be imported: a compiled dependency
# blocked by the operating system, a broken install, a shadowed
# package. Say so with the real cause instead of "not installed".
return ([_import_failed(exc)], False)
text = config if isinstance(config, str) else yaml.safe_dump(config, sort_keys=False)
with tempfile.TemporaryDirectory(prefix="pyxel-config-") as folder:
path = Path(folder) / "config.yaml"
# Bytes, not text mode: on Windows, text mode turns each line break
# into CRLF, so a file already written with CRLF reached Pyxel with
# every line break doubled.
path.write_bytes(text.encode("utf-8"))
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
try:
pyxel.load(path)
except Exception as exc: # noqa: BLE001 - every loader failure is a finding
failure = Diagnostic(
severity="error",
path="",
message=f"{type(exc).__name__}: {exc}",
hint="Reported by pyxel.load(), which is the final authority",
category="pyxel-load",
)
return [*_load_warnings(caught), failure], True
return _load_warnings(caught), True
[docs]
def validate_config(
config: dict[str, Any] | str, deep: bool = False, trusted: bool = False
) -> ValidationResult:
"""Validate against the bundled schema; `deep` also runs pyxel.load().
`trusted` lets the deep check run text Pyxel evaluates as Python (and
import user model modules) without reading it first; leave it False for
any config that did not come from you.
Parameters
----------
config : dict or str
The config as a mapping, or the raw YAML text of a ``.yaml`` file.
deep : bool, default False
Also hand the config to ``pyxel.load()`` (see `deep_check`), so what
Pyxel enforces only in its code is reported too. Needs Pyxel installed.
trusted : bool, default False
Let the deep check load the config without reading the values Pyxel
evaluates as Python first.
Returns
-------
ValidationResult
``valid`` is False when any finding is an error. Text that cannot be
read as a config is reported as a finding rather than raised.
Examples
--------
>>> from pyxel_config_core import validate_config
>>> result = validate_config(open("my_config.yaml").read())
>>> for d in result.diagnostics:
... print(d.severity, d.path, d.message)
"""
try:
data = _as_config(config)
except ConfigReadError as exc:
# Said as a finding, so a caller (an MCP client, the CLI's JSON) gets
# the reason rather than an exception.
return ValidationResult(valid=False, diagnostics=[read_problem(exc)])
# The schema, then what Pyxel enforces beyond it (run_rules.py). A focal
# plane has its own loader and no schema branch, so it is checked as the
# plane and its tiles (fpa.py).
if is_focal_plane(data):
diagnostics = _focal_plane_diagnostics(data)
else:
found = _schema_diagnostics(data)
diagnostics = (
found
+ run_rule_diagnostics(data, include_text_numbers=False)
+ _text_numbers_refused(
data, lambda c: [d.path for d in found] if c is data else _flat_schema_errors(c)
)
)
deep_checked = False
deep_version: str | None = None
if deep:
found, deep_checked = deep_check(config, trusted=trusted)
diagnostics.extend(found)
if deep_checked:
# The loader that just ran; it is imported by now.
deep_version = getattr(sys.modules.get("pyxel"), "__version__", None)
valid = not any(d.severity == "error" for d in diagnostics)
return ValidationResult(
valid=valid,
diagnostics=diagnostics,
deep_checked=deep_checked,
deep_pyxel_version=deep_version,
)
def _where(diagnostics: Iterable[Diagnostic]) -> list[str]:
"""The distinct places the errors live, in plain words."""
places: list[str] = []
for d in diagnostics:
parts = [p for p in d.path.split("/") if p]
if not parts:
place = "the top level"
elif parts[0] == FPA_KEY and len(parts) > 2 and parts[1] == "detectors":
place = f"tile {parts[2]}"
elif parts[0] == FPA_KEY:
place = "the focal plane"
elif parts[0] == "pipeline" and len(parts) > 1:
place = f"the {parts[1]} stage"
else:
place = f"the {parts[0]} block"
if place not in places:
places.append(place)
return places
def _loader_name(result: ValidationResult) -> str:
"""The Pyxel whose loader judged the config: "Pyxel 3.0.2"."""
return f"Pyxel {result.deep_pyxel_version}" if result.deep_pyxel_version else "Pyxel"
def _plane_summary(data: dict[str, Any], result: ValidationResult) -> str:
"""The valid sentence for a focal plane, which has no single schema branch."""
count = len(tiles(data))
sentence = (
f"This focal plane config is valid: each of its {count} "
f"tile{'s' if count != 1 else ''} passes the Pyxel {result.pyxel_version} schema "
"and the plane passes the checks of Pyxel's focal-plane loader"
)
if result.deep_checked:
sentence += f", and it loads in {_loader_name(result)}"
warnings = [d for d in result.diagnostics if d.severity == "warning"]
if warnings:
sentence += f", with {len(warnings)} warning{'s' if len(warnings) != 1 else ''}"
return sentence + "."
def _summary(data: dict[str, Any], result: ValidationResult) -> str:
plane = is_focal_plane(data)
if plane and result.valid:
return _plane_summary(data, result)
diagnosis = diagnose_shape(data)
detector = diagnosis.detector_key
mode = diagnosis.mode_key
kind = " ".join(
part
for part in (
detector.removesuffix("_detector").upper() if detector else "",
mode or "",
)
if part
)
if plane:
subject = "This focal plane config"
else:
subject = f"This {kind} config" if kind else "This config"
errors = [d for d in result.diagnostics if d.severity == "error"]
warnings = [d for d in result.diagnostics if d.severity == "warning"]
if not errors:
sentence = f"{subject} is valid against the Pyxel {result.pyxel_version} schema"
if result.deep_checked:
sentence += f" and loads in {_loader_name(result)}"
if warnings:
sentence += f", with {len(warnings)} warning{'s' if len(warnings) != 1 else ''}"
return sentence + "."
count = f"{len(errors)} problem{'s' if len(errors) != 1 else ''}"
places = _where(errors)
if len(places) == 1:
where = f" in {places[0]}"
elif len(places) <= 3:
where = " in " + ", ".join(places[:-1]) + f" and {places[-1]}"
else:
where = f" across {len(places)} places"
return f"{subject} has {count}{where}."
def _with_next_steps(data: dict[str, Any], diagnostics: list[Diagnostic]) -> list[Diagnostic]:
findings = {f.path: f for f in find_unknown_funcs(data)}
out: list[Diagnostic] = []
for d in diagnostics:
if d.next_step is None and d.path in findings:
d = d.model_copy(update={"next_step": next_step_for(findings[d.path])})
out.append(d)
return out
[docs]
def explain_config(
config: dict[str, Any] | str, deep: bool = False, trusted: bool = False
) -> Explanation:
"""validate_config, plus a one-sentence summary and a next step per finding.
Parameters
----------
config : dict or str
The config as a mapping, or the raw YAML text of a ``.yaml`` file.
deep : bool, default False
Also hand the config to ``pyxel.load()``, as `validate_config` does.
trusted : bool, default False
Let the deep check load the config without reading the values Pyxel
evaluates as Python first.
Returns
-------
Explanation
The same findings as `validate_config`, each with a ``next_step``,
under a ``summary`` a person can read first.
"""
try:
data = _as_config(config)
except ConfigReadError as exc:
problem = read_problem(exc)
return Explanation(
valid=False,
summary="This text could not be read as a Pyxel config, so it was not checked.",
diagnostics=[problem.model_copy(update={"next_step": problem.hint})],
)
result = validate_config(
data if isinstance(config, dict) else config, deep=deep, trusted=trusted
)
diagnostics = _with_next_steps(data, result.diagnostics)
return Explanation(
valid=result.valid,
summary=_summary(data, result),
diagnostics=diagnostics,
pyxel_version=result.pyxel_version,
deep_checked=result.deep_checked,
deep_pyxel_version=result.deep_pyxel_version,
)
