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
6 changes: 5 additions & 1 deletion radios/baofeng_dm32/capabilities.json
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,11 @@
"max_scan_lists": 32,
"max_channels_per_scan_list": 15,
"max_rx_group_lists": 32,
"max_talkgroups_per_rx_group_list": 32
"max_talkgroups_per_rx_group_list": 32,
"max_channel_name_chars": 16,
"max_zone_name_chars": 16,
"max_scan_list_name_chars": 10,
"max_contact_name_chars": 16
},
"bands": [
{
Expand Down
20 changes: 20 additions & 0 deletions schemas/capabilities-0.2.schema.json
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,26 @@
"type": "integer",
"minimum": 0,
"description": "Distinct DMR radio IDs the radio can store."
},
"max_channel_name_chars": {
"type": "integer",
"minimum": 1,
"description": "Characters a channel display name can hold. Names longer than this may be silently truncated by the CPS or on-radio encoder rather than rejected. Omit when unknown."
},
"max_zone_name_chars": {
"type": "integer",
"minimum": 1,
"description": "Characters a zone name can hold. Omit when unknown."
},
"max_scan_list_name_chars": {
"type": "integer",
"minimum": 1,
"description": "Characters a scan list name can hold. Omit when unknown."
},
"max_contact_name_chars": {
"type": "integer",
"minimum": 1,
"description": "Characters a contact name can hold. Omit when unknown."
}
}
},
Expand Down
66 changes: 49 additions & 17 deletions src/codeplugger/profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,28 @@ def _assignment_channel_counts(
return channel_counts, ambiguous


def _check_name_length(
limits: Mapping[str, int],
limit_key: str,
kind: str,
name: str,
) -> None:
"""Raise when a name exceeds the radio's storage for that name field.

Skipped when the capability is absent, so an incomplete capabilities file
degrades to today's behavior instead of producing false failures.
"""

limit = limits.get(limit_key)
if limit is None:
return
if len(name) > limit:
raise ProfileValidationError(
f"{kind} name '{name}' is {len(name)} characters; "
f"radio limit is {limit}"
)


def _band_label(band: Mapping[str, Any]) -> str:
name = band.get("name")
span = f"{band['min_mhz']}-{band['max_mhz']} MHz"
Expand Down Expand Up @@ -161,6 +183,31 @@ def _check_radio_support(
)


def _load_capabilities(
radio_id: str,
radio_root: Path = DEFAULT_RADIO_ROOT,
) -> dict[str, Any]:
"""Load and schema-validate a radio's capabilities document."""

radio_path = radio_root / radio_id / "capabilities.json"
if not radio_path.is_file():
raise ProfileValidationError(f"unknown radio '{radio_id}'")
capabilities = json.loads(radio_path.read_text(encoding="utf-8"))
if DEFAULT_CAPABILITIES_SCHEMA_PATH.is_file():
capabilities_schema = json.loads(
DEFAULT_CAPABILITIES_SCHEMA_PATH.read_text(encoding="utf-8")
)
capability_errors = sorted(
Draft202012Validator(capabilities_schema).iter_errors(capabilities),
key=lambda error: list(error.path),
)
if capability_errors:
raise ProfileValidationError(
f"{radio_path}: {_format_schema_errors(capability_errors)}"
)
return capabilities


def _load_and_validate_profile(
profile_path: Path,
ssrf_roots: Sequence[Path],
Expand All @@ -181,23 +228,7 @@ def _load_and_validate_profile(
f"{profile_path}: {_format_schema_errors(schema_errors)}"
)

radio_id = profile["radio"]
radio_path = radio_root / radio_id / "capabilities.json"
if not radio_path.is_file():
raise ProfileValidationError(f"unknown radio '{radio_id}'")
capabilities = json.loads(radio_path.read_text(encoding="utf-8"))
if DEFAULT_CAPABILITIES_SCHEMA_PATH.is_file():
capabilities_schema = json.loads(
DEFAULT_CAPABILITIES_SCHEMA_PATH.read_text(encoding="utf-8")
)
capability_errors = sorted(
Draft202012Validator(capabilities_schema).iter_errors(capabilities),
key=lambda error: list(error.path),
)
if capability_errors:
raise ProfileValidationError(
f"{radio_path}: {_format_schema_errors(capability_errors)}"
)
capabilities = _load_capabilities(profile["radio"], radio_root)

try:
from ssrf import resolve_ssrf_roots
Expand All @@ -223,6 +254,7 @@ def _load_and_validate_profile(
if zone["id"] in zone_ids:
raise ProfileValidationError(f"duplicate zone ID '{zone['id']}'")
zone_ids.add(zone["id"])
_check_name_length(limits, "max_zone_name_chars", "zone", zone["name"])
zone_channel_count = 0
for assignment_id in zone["assignments"]:
if assignment_id in ambiguous_assignments:
Expand Down
16 changes: 15 additions & 1 deletion src/codeplugger/resolved.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,13 @@

import yaml

from .profile import DEFAULT_RADIO_ROOT, DEFAULT_SCHEMA_PATH, _load_and_validate_profile
from .profile import (
DEFAULT_RADIO_ROOT,
DEFAULT_SCHEMA_PATH,
_check_name_length,
_load_and_validate_profile,
_load_capabilities,
)


@dataclass(frozen=True)
Expand Down Expand Up @@ -193,6 +199,7 @@ def resolve_codeplug(
schema_path=schema_path,
radio_root=radio_root,
)
limits = _load_capabilities(profile["radio"], radio_root)["limits"]
assignments = {
assignment.id: (document, assignment)
for document in documents
Expand All @@ -206,6 +213,13 @@ def resolve_codeplug(
for assignment_id in zone["assignments"]:
document, assignment = assignments[assignment_id]
resolved_channels = _resolve_assignment(document, assignment)
for resolved_channel in resolved_channels:
_check_name_length(
limits,
"max_channel_name_chars",
"channel",
resolved_channel.display_name,
)
channels.extend(resolved_channels)
channel_references.extend(
channel.reference for channel in resolved_channels
Expand Down
114 changes: 113 additions & 1 deletion tests/test_profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -326,4 +326,116 @@ def test_resolved_codeplug_output_is_repeatable() -> None:

assert first.to_json() == second.to_json()
assert first.to_yaml() == second.to_yaml()
assert json.loads(first.to_json()) == yaml.safe_load(first.to_yaml())
assert json.loads(first.to_json()) == yaml.safe_load(first.to_yaml())

def _write_radio_with_limits(root: Path, extra_limits: dict) -> None:
"""Write a test radio whose capabilities carry additional limit keys."""

radio = root / "test_radio"
radio.mkdir(parents=True, exist_ok=True)
limits = {
"max_channels": 2,
"max_zones": 1,
"max_channels_per_zone": 2,
}
limits.update(extra_limits)
(radio / "capabilities.json").write_text(
json.dumps({"id": "test_radio", "name": "Test radio", "limits": limits}),
encoding="utf-8",
)


def test_profile_enforces_zone_name_length() -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = root / "profile.yml"
_write_profile(profile, ["asg_one"])
data = yaml.safe_load(profile.read_text(encoding="utf-8"))
data["zones"][0]["name"] = "A" * 17
profile.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
_write_radio_with_limits(root / "radios", {"max_zone_name_chars": 16})
_write_ssrf(root / "ssrf")

with pytest.raises(ProfileValidationError, match="17 characters"):
load_and_validate_profile(
profile,
[root / "ssrf"],
radio_root=root / "radios",
)


def test_zone_name_at_limit_is_accepted() -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = root / "profile.yml"
_write_profile(profile, ["asg_one"])
data = yaml.safe_load(profile.read_text(encoding="utf-8"))
data["zones"][0]["name"] = "A" * 16
profile.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
_write_radio_with_limits(root / "radios", {"max_zone_name_chars": 16})
_write_ssrf(root / "ssrf")

loaded = load_and_validate_profile(
profile,
[root / "ssrf"],
radio_root=root / "radios",
)

assert loaded["zones"][0]["name"] == "A" * 16


def test_absent_name_limit_skips_check() -> None:
"""An incomplete capabilities file must degrade, not fail."""

with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = root / "profile.yml"
_write_profile(profile, ["asg_one"])
data = yaml.safe_load(profile.read_text(encoding="utf-8"))
data["zones"][0]["name"] = "A" * 200
profile.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
_write_radio_with_limits(root / "radios", {})
_write_ssrf(root / "ssrf")

loaded = load_and_validate_profile(
profile,
[root / "ssrf"],
radio_root=root / "radios",
)

assert loaded["zones"][0]["name"] == "A" * 200


def test_resolver_enforces_channel_name_length() -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
profile = root / "profile.yml"
_write_profile(profile, ["asg_one"])
_write_radio_with_limits(root / "radios", {"max_channel_name_chars": 8})
_write_ssrf(root / "ssrf")
fixture = root / "ssrf" / "systems" / "fixture.yml"
data = yaml.safe_load(fixture.read_text(encoding="utf-8"))
data["assignments"][0]["display_name"] = "WAY TOO LONG NAME"
fixture.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")

with pytest.raises(ProfileValidationError, match="channel name"):
resolve_codeplug(
profile,
[root / "ssrf"],
radio_root=root / "radios",
)


def test_dm32_capabilities_declare_name_limits() -> None:
"""The shipped DM-32 document must carry the field-verified name limits."""

capabilities = json.loads(
(Path(__file__).resolve().parents[1] / "radios" / "baofeng_dm32" / "capabilities.json").read_text(
encoding="utf-8"
)
)
limits = capabilities["limits"]
assert limits["max_channel_name_chars"] == 16
assert limits["max_zone_name_chars"] == 16
assert limits["max_scan_list_name_chars"] == 10
assert limits["max_contact_name_chars"] == 16
Loading