Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -187,3 +187,33 @@ Notes:
- Any non-FM channel selected by the profile fails export with an explicit
error.
- Channel order in the CSV matches resolved profile order.

## Optional instance registry

Profiles may declare `radio_instance`, and codeplugger can optionally validate
that identifier against a separate instance registry document:

```yaml
version: "0.1"
instances:
dm32_green_01:
radio: baofeng_dm32
label: "Green DM-32"
firmware: "DM32.01.L01.048"
```

Use `--instance-registry` to enable this join check:

```bash
uv run codeplugger-profile path/to/profile.yml \
--ssrf-root ../ssrf-lite/ssrf \
--instance-registry path/to/instances.yml
```

When enabled, codeplugger verifies:

- the profile instance ID exists in the registry
- the registry entry's `radio` matches the profile's `radio`

If no registry is provided, behavior stays backward compatible with existing
profile-only workflows.
40 changes: 40 additions & 0 deletions schemas/instance-registry-0.1.schema.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://raw.githubusercontent.com/Chicago-Offline/codeplugger/main/schemas/instance-registry-0.1.schema.json",
"title": "Codeplugger instance registry 0.1",
"type": "object",
"additionalProperties": false,
"required": ["version", "instances"],
"properties": {
"$schema": { "type": "string" },
"version": { "const": "0.1" },
"instances": {
"type": "object",
"minProperties": 1,
"propertyNames": {
"pattern": "^[a-z0-9][a-z0-9_-]*$"
},
"additionalProperties": {
"type": "object",
"required": ["radio"],
"properties": {
"radio": {
"type": "string",
"pattern": "^[a-z0-9][a-z0-9_-]*$"
},
"profile": { "type": "string" },
"label": { "type": "string" },
"owner": { "type": "string" },
"case_serial": { "type": "string" },
"fcc_id": { "type": "string" },
"hardware_family": { "type": "string" },
"firmware": { "type": "string" },
"sk_buttons": {
"type": ["string", "null"]
}
},
"additionalProperties": true
}
}
}
}
73 changes: 70 additions & 3 deletions src/codeplugger/profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,9 @@
DEFAULT_CAPABILITIES_SCHEMA_PATH = (
PROJECT_ROOT / "schemas" / "capabilities-0.2.schema.json"
)
DEFAULT_INSTANCE_REGISTRY_SCHEMA_PATH = (
PROJECT_ROOT / "schemas" / "instance-registry-0.1.schema.json"
)
DEFAULT_RADIO_ROOT = PROJECT_ROOT / "radios"


Expand Down Expand Up @@ -208,13 +211,59 @@ def _load_capabilities(
return capabilities


def _load_instance_registry(
registry_path: Path,
*,
schema_path: Path = DEFAULT_INSTANCE_REGISTRY_SCHEMA_PATH,
) -> dict[str, Any]:
"""Load and schema-validate an instance registry document."""

registry = _load_mapping(registry_path)
if schema_path.is_file():
schema = json.loads(schema_path.read_text(encoding="utf-8"))
errors = sorted(
Draft202012Validator(schema).iter_errors(registry),
key=lambda error: list(error.path),
)
if errors:
raise ProfileValidationError(
f"{registry_path}: {_format_schema_errors(errors)}"
)
return registry


def _validate_instance_reference(
profile: Mapping[str, Any],
registry_path: Path,
registry: Mapping[str, Any],
) -> dict[str, Any]:
"""Ensure the profile's radio_instance resolves to matching registry data."""

instances = registry.get("instances", {})
instance_id = profile.get("radio_instance", profile["id"])
if instance_id not in instances:
raise ProfileValidationError(
f"{registry_path}: missing instance '{instance_id}'"
)

instance = instances[instance_id]
instance_radio = instance.get("radio")
if instance_radio != profile["radio"]:
raise ProfileValidationError(
f"{registry_path}: instance '{instance_id}' targets radio "
f"'{instance_radio}', but profile targets '{profile['radio']}'"
)
return dict(instance)


def _load_and_validate_profile(
profile_path: Path,
ssrf_roots: Sequence[Path],
*,
schema_path: Path = DEFAULT_SCHEMA_PATH,
radio_root: Path = DEFAULT_RADIO_ROOT,
) -> tuple[dict[str, Any], Sequence[Any]]:
instance_registry_path: Path | None = None,
) -> tuple[dict[str, Any], Sequence[Any], dict[str, Any] | None]:
"""Load a profile and its validated, overlay-resolved SSRF documents."""

profile = _load_mapping(profile_path)
Expand All @@ -229,6 +278,14 @@ def _load_and_validate_profile(
)

capabilities = _load_capabilities(profile["radio"], radio_root)
instance_metadata: dict[str, Any] | None = None
if instance_registry_path is not None:
registry = _load_instance_registry(instance_registry_path)
instance_metadata = _validate_instance_reference(
profile,
instance_registry_path,
registry,
)

try:
from ssrf import resolve_ssrf_roots
Expand Down Expand Up @@ -287,7 +344,7 @@ def _load_and_validate_profile(
)

_check_radio_support(capabilities, documents, selected)
return profile, documents
return profile, documents, instance_metadata


def load_and_validate_profile(
Expand All @@ -296,14 +353,16 @@ def load_and_validate_profile(
*,
schema_path: Path = DEFAULT_SCHEMA_PATH,
radio_root: Path = DEFAULT_RADIO_ROOT,
instance_registry_path: Path | None = None,
) -> dict[str, Any]:
"""Load a profile and validate its schema, references, and radio limits."""

profile, _ = _load_and_validate_profile(
profile, _, _ = _load_and_validate_profile(
profile_path,
ssrf_roots,
schema_path=schema_path,
radio_root=radio_root,
instance_registry_path=instance_registry_path,
)
return profile

Expand All @@ -319,6 +378,12 @@ def main() -> int:
help="SSRF root in precedence order; repeat for overlays",
)
parser.add_argument("--radio-root", type=Path, default=DEFAULT_RADIO_ROOT)
parser.add_argument(
"--instance-registry",
type=Path,
default=None,
help="optional path to instance registry file",
)
parser.add_argument(
"--output-format",
choices=("summary", "json", "yaml", "chirp-csv"),
Expand All @@ -333,6 +398,7 @@ def main() -> int:
args.profile,
args.ssrf_root,
radio_root=args.radio_root,
instance_registry_path=args.instance_registry,
)
else:
from .resolved import resolve_codeplug
Expand All @@ -341,6 +407,7 @@ def main() -> int:
args.profile,
args.ssrf_root,
radio_root=args.radio_root,
instance_registry_path=args.instance_registry,
)
except (OSError, ProfileValidationError, ValueError) as exc:
parser.exit(1, f"error: {exc}\n")
Expand Down
6 changes: 5 additions & 1 deletion src/codeplugger/resolved.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ class ResolvedCodeplug:

radio_id: str
radio_instance_id: str
radio_instance: dict[str, Any] | None
channels: tuple[ResolvedChannel, ...]
zones: tuple[ResolvedZone, ...]

Expand Down Expand Up @@ -190,14 +191,16 @@ def resolve_codeplug(
*,
schema_path: Path = DEFAULT_SCHEMA_PATH,
radio_root: Path = DEFAULT_RADIO_ROOT,
instance_registry_path: Path | None = None,
) -> ResolvedCodeplug:
"""Validate and normalize a profile plus precedence-ordered SSRF roots."""

profile, documents = _load_and_validate_profile(
profile, documents, instance_metadata = _load_and_validate_profile(
profile_path,
ssrf_roots,
schema_path=schema_path,
radio_root=radio_root,
instance_registry_path=instance_registry_path,
)
limits = _load_capabilities(profile["radio"], radio_root)["limits"]
assignments = {
Expand Down Expand Up @@ -235,6 +238,7 @@ def resolve_codeplug(
return ResolvedCodeplug(
radio_id=profile["radio"],
radio_instance_id=profile.get("radio_instance", profile["id"]),
radio_instance=instance_metadata,
channels=tuple(channels),
zones=tuple(zones),
)
1 change: 1 addition & 0 deletions tests/test_chirp_csv.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ def _codeplug(channels: list[ResolvedChannel]) -> ResolvedCodeplug:
return ResolvedCodeplug(
radio_id="baofeng_uv5r_mini",
radio_instance_id="5rm_01",
radio_instance=None,
channels=tuple(channels),
zones=(ResolvedZone(id="z1", name="Reference", channel_references=refs),),
)
Expand Down
99 changes: 99 additions & 0 deletions tests/test_profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,21 @@ def _write_ssrf(root: Path) -> None:
)


def _write_instance_registry(root: Path, instances: dict[str, dict]) -> Path:
path = root / "instances.yml"
path.write_text(
yaml.safe_dump(
{
"version": "0.1",
"instances": instances,
},
sort_keys=False,
),
encoding="utf-8",
)
return path


def test_profile_resolves_ordered_assignment_ids() -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
Expand Down Expand Up @@ -141,6 +156,7 @@ def test_resolved_codeplug_defaults_instance_to_profile_id() -> None:
)

assert resolved.radio_instance_id == "test_profile"
assert resolved.radio_instance is None


def test_profile_rejects_unknown_assignment() -> None:
Expand Down Expand Up @@ -498,3 +514,86 @@ def test_uv5r_mini_fixture_profile_validates_end_to_end() -> None:

assert loaded["radio"] == "baofeng_uv5r_mini"
assert loaded["zones"][0]["assignments"] == ["asg_one"]


def test_profile_rejects_missing_instance_registry_entry() -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = root / "profile.yml"
_write_profile(profile, ["asg_one"])
_write_radio(root / "radios")
_write_ssrf(root / "ssrf")
registry = _write_instance_registry(
root,
{
"other_radio_01": {
"radio": "test_radio",
}
},
)

with pytest.raises(ProfileValidationError, match="missing instance"):
load_and_validate_profile(
profile,
[root / "ssrf"],
radio_root=root / "radios",
instance_registry_path=registry,
)


def test_profile_rejects_instance_registry_radio_mismatch() -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = root / "profile.yml"
_write_profile(profile, ["asg_one"])
_write_radio(root / "radios")
_write_ssrf(root / "ssrf")
registry = _write_instance_registry(
root,
{
"dm32_green_01": {
"radio": "some_other_radio",
}
},
)

with pytest.raises(ProfileValidationError, match="targets radio"):
load_and_validate_profile(
profile,
[root / "ssrf"],
radio_root=root / "radios",
instance_registry_path=registry,
)


def test_resolved_codeplug_includes_instance_registry_metadata() -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = root / "profile.yml"
_write_profile(profile, ["asg_one"])
_write_radio(root / "radios")
_write_ssrf(root / "ssrf")
registry = _write_instance_registry(
root,
{
"dm32_green_01": {
"radio": "test_radio",
"label": "Green test radio",
"firmware": "TEST.01",
}
},
)

resolved = resolve_codeplug(
profile,
[root / "ssrf"],
radio_root=root / "radios",
instance_registry_path=registry,
)

assert resolved.radio_instance_id == "dm32_green_01"
assert resolved.radio_instance == {
"radio": "test_radio",
"label": "Green test radio",
"firmware": "TEST.01",
}
Loading