plesty.lib.test.schema_refresh

Turn field-test observations into a reviewed parameter-schema update.

A field test already reads every parameter and asks the instrument for its own limits. That is precisely the information a hand-written schema drifts away from — the shipped envelope is what the console can do, while the attached sensor is usually narrower.

This module turns those observations into a proposal, never a silent edit. Nothing is written to a schema without a developer saying yes, because the three kinds of change carry very different confidence:

type: Ground truth. The instrument’s response type is what the parser must handle, and a mismatch is a real defect.
min_value / max_value / options: The attached sensor’s actual limits, which is what a schema envelope should describe. High confidence, and usually the whole point.
default: Advisory only. The value read is whatever the current experiment set, not a sensible shipping default. Writing “wavelength = 780 nm” into a module’s schema because one measurement happened to use 780 nm makes an experiment-specific value permanent for every future user. Proposed so it is visible, flagged so it is not applied by reflex.

Typical usage:

proposal = pipeline.propose_schema_update()
if proposal.has_changes:
    print(proposal.to_markdown())
    proposal.write("reports/schema-proposed")   # never touches the original
    proposal.apply(confirm=confirm_interactively)   # asks first

Attributes

`TRUSTED_FIELDS`
`ADVISORY_FIELDS`
`_TYPE_NAMES`

Classes

`SchemaChange`	One proposed difference between a schema entry and the instrument.
`SchemaProposal`	Proposed schema changes derived from what an instrument reported.

Functions

`confirm_interactively`(→ bool)	Print the diff and ask on the terminal whether to apply it.
`propose_schema_update`(→ SchemaProposal)	Build a schema proposal from what an instrument reported.
`_iter_entries`(→ collections.abc.Iterator[tuple[str, ...)	Yield `(runtime_key, entry)` for every parameter, entries mutable.
`_schema_range`(→ tuple[Any, Any])	Return the `(min, max)` a schema entry declares, in either spelling.
`_set_field`(→ None)	Write one proposed field back into a schema entry, keeping its shape.

Module Contents

plesty.lib.test.schema_refresh.TRUSTED_FIELDS: tuple[str, Ellipsis] = ('type', 'min_value', 'max_value', 'options')

plesty.lib.test.schema_refresh.ADVISORY_FIELDS: tuple[str, Ellipsis] = ('default',)

plesty.lib.test.schema_refresh._TYPE_NAMES: dict[type, str]

class plesty.lib.test.schema_refresh.SchemaChange

One proposed difference between a schema entry and the instrument.

Variables:

key – Runtime parameter key.
field – Schema field the change applies to.
current – Value currently in the schema.
observed – Value the instrument reported.
advisory – True when the change needs judgement rather than acceptance — see ADVISORY_FIELDS.
note – Why this change is proposed, or why it needs care.

key: str

field: str

current: Any

observed: Any

advisory: bool

note: str = ''

to_dict() → dict[str, Any]

Return a JSON-serialisable mapping of the change.

Return type:: dict[str, Any]

class plesty.lib.test.schema_refresh.SchemaProposal

Proposed schema changes derived from what an instrument reported.

Variables:

schema_path – The schema file the proposal was built against.
device – Device class name.
address – Instrument address the observations came from.
observed_utc – When the observations were taken.
changes – Every proposed change, trusted ones first.

schema_path: str

device: str = ''

address: str = ''

observed_utc: str = ''

changes: list[SchemaChange] = []

property has_changes: bool

True when the instrument disagreed with the schema anywhere.

Return type:: bool

property trusted: list[SchemaChange]

Changes the instrument is authoritative about.

Return type:: list[SchemaChange]

property advisory: list[SchemaChange]

Changes that need a developer’s judgement.

Return type:: list[SchemaChange]

to_dict() → dict[str, Any]

Return a JSON-serialisable mapping of the proposal.

Return type:: dict[str, Any]

to_markdown() → str

Render the proposal as a reviewable diff.

Return type:: str

updated_schema(include_advisory: bool = False) → dict[str, Any]

Return the schema with the proposed changes applied in memory.

The file on disk is untouched.

Parameters:: include_advisory (bool) – Also apply the changes flagged for review.
Return type:: dict[str, Any]

write(path: str | pathlib.Path, include_advisory: bool = False) → list[pathlib.Path]

Write the proposal beside the schema without modifying the original.

Parameters:

path (str | pathlib.Path) – Base path; writes <base>.json (the full updated schema) and <base>.md (the reviewable diff).
include_advisory (bool) – Also apply the changes flagged for review.

Returns:

The paths written, schema first.

Return type:

list[pathlib.Path]

apply(confirm: collections.abc.Callable[[str], bool], include_advisory: bool = False) → bool

Overwrite the schema in place, but only if confirm returns True.

There is deliberately no way to apply without asking: a schema is a module’s published contract, and a reading taken during one experiment is not automatically the right thing to ship.

Parameters:

confirm (collections.abc.Callable[[str], bool]) – Called with the rendered diff; returns True to proceed. Callers on a terminal can pass confirm_interactively().
include_advisory (bool) – Also apply the changes flagged for review.

Returns:

True if the schema was rewritten, False if there was nothing to change or the developer declined.

Return type:

bool

plesty.lib.test.schema_refresh.confirm_interactively(diff: str) → bool

Print the diff and ask on the terminal whether to apply it.

Returns False without prompting when stdin is not a terminal, so an unattended run can never answer the question by accident.

Parameters:: diff (str) – The rendered proposal.
Return type:: bool

plesty.lib.test.schema_refresh.propose_schema_update(schema_path: str | pathlib.Path, observations: dict[str, dict[str, Any]], *, device: str = '', address: str = '') → SchemaProposal

Build a schema proposal from what an instrument reported.

Parameters:

schema_path (str | pathlib.Path) – Path to the parameter schema JSON.
observations (dict[str, dict[str, Any]]) – Per runtime key, any of value, min_value, max_value and options as read from the instrument.
device (str) – Device class name, recorded in the proposal.
address (str) – Instrument address, recorded in the proposal.

Returns:

The SchemaProposal; empty when the schema already agrees.

Return type:

SchemaProposal

plesty.lib.test.schema_refresh._iter_entries(schema: dict[str, Any]) → collections.abc.Iterator[tuple[str, dict[str, Any]]]

Yield (runtime_key, entry) for every parameter, entries mutable.

Handles the flat, dotted and grouped schema shapes, matching plesty.lib.test.device_pipeline. The yielded entry is the live dictionary from the loaded schema, so writing to it updates the schema.

Parameters:: schema (dict[str, Any]) – The loaded schema document.
Return type:: collections.abc.Iterator[tuple[str, dict[str, Any]]]

plesty.lib.test.schema_refresh._schema_range(entry: dict[str, Any]) → tuple[Any, Any]

Return the (min, max) a schema entry declares, in either spelling.

Parameters:: entry (dict[str, Any]) – One parameter entry.
Return type:: tuple[Any, Any]

plesty.lib.test.schema_refresh._set_field(entry: dict[str, Any], field_name: str, value: Any) → None

Write one proposed field back into a schema entry, keeping its shape.

Parameters:

entry (dict[str, Any]) – The parameter entry to update.
field_name (str) – One of the fields carried by SchemaChange.
value (Any) – The new value.

Return type:

None