Migrate parameters to methods-only command schema

Author: BB-84C

config/parameters.yaml

@@ -5,14 +5,14 @@ Expose a small, stable command surface that orchestration agents can call direct
 
 ## Core commands
 - `nqctl capabilities`: returns parameters/action-commands/actions/policy summary,
-  including rich `parameters.items[*]` metadata (`get_cmd`, `set_cmd`, `vals`,
+  including rich `parameters.items[*]` metadata (`get_cmd`, `set_cmd`,
   `safety`) and `action_commands.items[*]` metadata (`action_cmd`, `safety_mode`).
   Descriptions are exposed on command blocks when present.
 - `nqctl observables list`: returns readable and writable parameter metadata.
 - `nqctl actions list`: returns supported action descriptors.
 - `nqctl act <action_name> --arg <key=value>`: invoke one manifest action command.
 - `nqctl get <parameter>`: reads one parameter value.
-- `nqctl set <parameter> <value>`: guarded strict single-step write.
+- `nqctl set <parameter> --arg <key=value>`: guarded strict write using structured argument fields.
 - `nqctl ramp <parameter> <start> <end> <step> --interval-s <sec>`: guarded explicit ramp.
 - `nqctl policy show`: returns effective write policy and enablement guidance.
 

docs/cli_contract.md

@@ -619,7 +619,6 @@ def _cmd_get(args: argparse.Namespace) -> int:
         args, auto_connect=True, include_parameters=(parameter_name,)
     ) as instrument_ctx:
         instrument, journal = instrument_ctx
-        spec = instrument.parameter_spec(parameter_name)
         snapshot = instrument.get_parameter_snapshot(parameter_name)
         values = snapshot.get("values", {})
         if len(values) == 1:
@@ -630,7 +629,6 @@ def _cmd_get(args: argparse.Namespace) -> int:
             "parameter": parameter_name,
             "value": _json_safe(value),
             "fields": _json_safe(values),
-            "unit": spec.unit,
             "timestamp_utc": _now_utc_iso(),
         }
         if journal is not None:
@@ -842,7 +840,8 @@ def _cmd_parameters_validate(args: argparse.Namespace) -> int:
                 "name": spec.name,
                 "readable": spec.readable,
                 "writable": spec.writable,
-                "value_type": spec.value_type,
+                "has_get_cmd": spec.get_cmd is not None,
+                "has_set_cmd": spec.set_cmd is not None,
             }
             for spec in specs
         ],
@@ -1033,8 +1032,11 @@ def _cmd_trajectory_monitor_list_signals(args: argparse.Namespace) -> int:
         {
             "name": spec.name,
             "label": spec.label,
-            "unit": spec.unit,
-            "value_type": spec.value_type,
+            "response_fields": (
+                []
+                if spec.get_cmd is None
+                else [asdict(field) for field in spec.get_cmd.response_fields]
+            ),
         }
         for spec in specs
         if spec.readable
@@ -1051,9 +1053,23 @@ def _cmd_trajectory_monitor_list_specs(args: argparse.Namespace) -> int:
         {
             "name": spec.name,
             "label": spec.label,
-            "unit": spec.unit,
-            "value_type": spec.value_type,
-            "vals": None if spec.vals is None else asdict(spec.vals),
+            "get_cmd": (
+                None
+                if spec.get_cmd is None
+                else {
+                    "command": spec.get_cmd.command,
+                    "arg_fields": [asdict(field) for field in spec.get_cmd.arg_fields],
+                    "response_fields": [asdict(field) for field in spec.get_cmd.response_fields],
+                }
+            ),
+            "set_cmd": (
+                None
+                if spec.set_cmd is None
+                else {
+                    "command": spec.set_cmd.command,
+                    "arg_fields": [asdict(field) for field in spec.set_cmd.arg_fields],
+                }
+            ),
         }
         for spec in specs
         if spec.readable
@@ -1274,10 +1290,8 @@ def _collect_observables(instrument: Any) -> list[dict[str, Any]]:
             {
                 "name": spec.name,
                 "label": spec.label,
-                "unit": spec.unit,
                 "readable": spec.readable,
                 "writable": spec.writable,
-                "value_type": spec.value_type,
                 "has_ramp": bool(spec.safety is not None and spec.safety.ramp_enabled),
             }
         )
@@ -1318,15 +1332,6 @@ def _collect_parameter_capabilities(instrument: Any) -> list[dict[str, Any]]:
             if set_description is not None:
                 set_cmd["description"] = set_description
 
-        vals = None
-        if spec.vals is not None:
-            vals = {
-                "kind": spec.vals.kind,
-                "min_value": spec.vals.min_value,
-                "max_value": spec.vals.max_value,
-                "choices": list(spec.vals.choices),
-            }
-
         safety = None
         if spec.safety is not None:
             safety = {
@@ -1342,15 +1347,11 @@ def _collect_parameter_capabilities(instrument: Any) -> list[dict[str, Any]]:
         capability: dict[str, Any] = {
             "label": spec.label,
             "name": spec.name,
-            "unit": spec.unit,
-            "value_type": spec.value_type,
-            "snapshot_value": spec.snapshot_value,
             "readable": bool(spec.readable),
             "writable": bool(spec.writable),
             "has_ramp": bool(spec.safety is not None and spec.safety.ramp_enabled),
             "get_cmd": get_cmd,
             "set_cmd": set_cmd,
-            "vals": vals,
             "safety": safety,
         }
 

nanonis_qcodes_controller/cli.py

@@ -1,19 +1,15 @@
 from .extensions import (
     ActionSpec,
     ParameterSpec,
-    ScalarParameterSpec,
     load_action_specs,
     load_parameter_specs,
-    load_scalar_parameter_specs,
 )
 from .instrument import QcodesNanonisSTM
 
 __all__ = [
     "QcodesNanonisSTM",
     "ActionSpec",
     "ParameterSpec",
-    "ScalarParameterSpec",
     "load_action_specs",
     "load_parameter_specs",
-    "load_scalar_parameter_specs",
 ]

nanonis_qcodes_controller/qcodes_driver/__init__.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Literal, cast
 
@@ -12,11 +12,9 @@
 DEFAULT_PARAMETERS_FILE = Path("config/parameters.yaml")
 
 ScalarValueType = Literal["float", "int", "bool", "str"]
-ValidatorKind = Literal["numbers", "ints", "bool", "enum", "none"]
 ActionSafetyMode = Literal["alwaysAllowed", "guarded", "blocked"]
 
 _ALLOWED_VALUE_TYPES: frozenset[str] = frozenset({"float", "int", "bool", "str"})
-_ALLOWED_VALIDATOR_KINDS: frozenset[str] = frozenset({"numbers", "ints", "bool", "enum", "none"})
 _ALLOWED_ACTION_SAFETY_MODES: frozenset[str] = frozenset({"alwaysAllowed", "guarded", "blocked"})
 
 
@@ -72,14 +70,6 @@ class ActionSpec:
     safety_mode: ActionSafetyMode = "guarded"
 
 
-@dataclass(frozen=True)
-class ValidatorSpec:
-    kind: ValidatorKind
-    min_value: float | None = None
-    max_value: float | None = None
-    choices: tuple[Any, ...] = ()
-
-
 @dataclass(frozen=True)
 class SafetySpec:
     min_value: float | None
@@ -95,13 +85,9 @@ class SafetySpec:
 class ParameterSpec:
     name: str
     label: str
-    unit: str
-    value_type: ScalarValueType
     get_cmd: ReadCommandSpec | None
     set_cmd: WriteCommandSpec | None
-    vals: ValidatorSpec | None
     safety: SafetySpec | None
-    snapshot_value: bool = True
     description: str = ""
 
     @property
@@ -194,54 +180,31 @@ def _parse_parameter_spec(
     defaults: Mapping[str, Any],
 ) -> ParameterSpec:
     label = str(mapping.get("label", name)).strip() or name
-    unit = str(mapping.get("unit", "")).strip()
     description = str(mapping.get("description", "")).strip()
 
-    raw_value_type = mapping.get("value_type", mapping.get("type", "float"))
-    value_type_text = str(raw_value_type).strip().lower()
-    if value_type_text not in _ALLOWED_VALUE_TYPES:
-        allowed = ", ".join(sorted(_ALLOWED_VALUE_TYPES))
-        raise ValueError(
-            f"parameters.{name}.value_type must be one of: {allowed}. Received: {raw_value_type}"
-        )
-    value_type = cast(ScalarValueType, value_type_text)
-
     get_cmd = _parse_read_command(mapping.get("get_cmd"), context=f"parameters.{name}.get_cmd")
     set_cmd = _parse_write_command(mapping.get("set_cmd"), context=f"parameters.{name}.set_cmd")
 
     if get_cmd is None and set_cmd is None:
         raise ValueError(f"Parameter '{name}' must define at least one of get_cmd or set_cmd.")
 
-    vals = _parse_vals(
-        mapping.get("vals"), value_type=value_type, context=f"parameters.{name}.vals"
-    )
     safety = _parse_safety(
         mapping.get("safety"),
-        vals=vals,
         defaults=defaults,
         context=f"parameters.{name}.safety",
         writable=set_cmd is not None,
     )
 
-    snapshot_value = _parse_bool(
-        mapping.get("snapshot_value", defaults.get("snapshot_value", True)),
-        field_name=f"parameters.{name}.snapshot_value",
-    )
-
     if set_cmd is not None and safety is None:
         raise ValueError(f"Writable parameter '{name}' must include safety settings.")
 
     return ParameterSpec(
         name=name,
         label=label,
-        unit=unit,
         description=description,
-        value_type=value_type,
         get_cmd=get_cmd,
         set_cmd=set_cmd,
-        vals=vals,
         safety=safety,
-        snapshot_value=snapshot_value,
     )
 
 
@@ -479,44 +442,9 @@ def _legacy_arg_fields_from_mapping(
     return tuple(fields)
 
 
-def _parse_vals(value: Any, *, value_type: ScalarValueType, context: str) -> ValidatorSpec | None:
-    if value is None:
-        if value_type == "bool":
-            return ValidatorSpec(kind="bool")
-        return None
-    if value is False:
-        return None
-
-    mapping = _as_mapping(value, context=context)
-    kind_raw = mapping.get("kind", _default_validator_kind(value_type))
-    kind = str(kind_raw).strip().lower()
-    if kind not in _ALLOWED_VALIDATOR_KINDS:
-        allowed = ", ".join(sorted(_ALLOWED_VALIDATOR_KINDS))
-        raise ValueError(f"{context}.kind must be one of: {allowed}. Received: {kind_raw}")
-
-    min_value = None if mapping.get("min") is None else float(mapping["min"])
-    max_value = None if mapping.get("max") is None else float(mapping["max"])
-    if min_value is not None and max_value is not None and max_value < min_value:
-        raise ValueError(f"{context}: max must be >= min.")
-
-    choices_raw = mapping.get("choices", ())
-    if isinstance(choices_raw, (list, tuple)):
-        choices = tuple(choices_raw)
-    else:
-        raise ValueError(f"{context}.choices must be a list when provided.")
-
-    return ValidatorSpec(
-        kind=cast(ValidatorKind, kind),
-        min_value=min_value,
-        max_value=max_value,
-        choices=choices,
-    )
-
-
 def _parse_safety(
     value: Any,
     *,
-    vals: ValidatorSpec | None,
     defaults: Mapping[str, Any],
     context: str,
     writable: bool,
@@ -531,8 +459,8 @@ def _parse_safety(
     if not writable and not mapping:
         return None
 
-    min_default = vals.min_value if vals is not None else None
-    max_default = vals.max_value if vals is not None else None
+    min_default = None
+    max_default = None
     max_step_default = None
 
     min_value_raw = mapping.get("min", min_default)
@@ -589,16 +517,6 @@ def _parse_safety(
     )
 
 
-def _default_validator_kind(value_type: ScalarValueType) -> ValidatorKind:
-    if value_type == "float":
-        return "numbers"
-    if value_type == "int":
-        return "ints"
-    if value_type == "bool":
-        return "bool"
-    return "none"
-
-
 def _parse_required_string(value: Any, *, field_name: str) -> str:
     text = "" if value is None else str(value).strip()
     if not text:
@@ -627,70 +545,3 @@ def _parse_bool(value: Any, *, field_name: str) -> bool:
         return False
 
     raise ValueError(f"Invalid boolean value for {field_name}: {value}")
-
-
-@dataclass(frozen=True)
-class ScalarParameterSpec:
-    name: str
-    command: str
-    value_type: ScalarValueType = "float"
-    unit: str = ""
-    label: str | None = None
-    payload_index: int = 0
-    args: Mapping[str, Any] = field(default_factory=dict)
-    snapshot_value: bool = True
-
-
-def load_scalar_parameter_specs(parameter_file: str | Path) -> tuple[ScalarParameterSpec, ...]:
-    parameter_path = Path(parameter_file).expanduser()
-    if not parameter_path.exists():
-        raise ValueError(f"Parameter file does not exist: {parameter_path}")
-
-    with parameter_path.open("r", encoding="utf-8") as handle:
-        loaded = yaml.safe_load(handle)
-
-    if loaded is None:
-        return ()
-
-    root = _as_mapping(loaded, context="root")
-    parameters_raw = root.get("parameters", [])
-    if not isinstance(parameters_raw, list):
-        raise ValueError("Parameter file field 'parameters' must be a list.")
-
-    specs: list[ScalarParameterSpec] = []
-    for index, entry in enumerate(parameters_raw):
-        context = f"parameters[{index}]"
-        mapping = _as_mapping(entry, context=context)
-
-        raw_value_type = mapping.get("value_type", mapping.get("type", "float"))
-        value_type_text = str(raw_value_type).strip().lower()
-        if value_type_text not in _ALLOWED_VALUE_TYPES:
-            allowed = ", ".join(sorted(_ALLOWED_VALUE_TYPES))
-            raise ValueError(
-                f"{context}.value_type must be one of: {allowed}. Received: {raw_value_type}"
-            )
-
-        payload_index = int(mapping.get("payload_index", 0))
-        if payload_index < 0:
-            raise ValueError(f"{context}.payload_index must be non-negative.")
-
-        args = _as_mapping(mapping.get("args"), context=f"{context}.args")
-        specs.append(
-            ScalarParameterSpec(
-                name=_parse_required_string(mapping.get("name"), field_name=f"{context}.name"),
-                command=_parse_required_string(
-                    mapping.get("command"), field_name=f"{context}.command"
-                ),
-                value_type=cast(ScalarValueType, value_type_text),
-                unit=str(mapping.get("unit", "")).strip(),
-                label=(None if mapping.get("label") is None else str(mapping.get("label")).strip()),
-                payload_index=payload_index,
-                args=dict(args),
-                snapshot_value=_parse_bool(
-                    mapping.get("snapshot_value", True),
-                    field_name=f"{context}.snapshot_value",
-                ),
-            )
-        )
-
-    return tuple(specs)