#!/usr/bin/env python3 """Validate an OCETS v1.1 Registry (OCETS-REGISTRY-1.1 Sections 12-13).""" import json import sys from collections import defaultdict from datetime import datetime from pathlib import Path import yaml try: import jsonschema except ImportError: jsonschema = None HERE = Path(__file__).resolve().parent DEFAULT_REGISTRY = HERE.parent / "examples" / "solar-plant-example.yaml" SCHEMA_PATH = HERE.parent / "schema" / "plant-registry-v1.1.schema.json" CATALOGUE_METRICS = { "active_power", "reactive_power", "apparent_power", "power_factor", "frequency", "voltage", "current", "energy_import", "energy_export", "energy_generated", "energy_consumed", "energy_charged", "energy_discharged", "state_of_charge", "state_of_health", "charge_limit", "discharge_limit", "temperature", "ambient_temperature", "irradiance", "wind_speed", "device_available", "communication_status", "collector_status", "data_freshness", "timestamp_skew", "mapping_error_count", "validation_error_count", "pipeline_status", "time_sync_status", "buffer_utilization", "buffer_oldest_age", "storage_utilization", "storage_status", "transport_error_count", "collector_restart_count", "cpu_utilization", "memory_utilization", "host_temperature", "external_power_available", "backup_power_status", "backup_power_runtime", } EDGE_PROFILE_REQUIRED = { "device_available", "collector_status", "pipeline_status", "communication_status", "data_freshness", "time_sync_status", "timestamp_skew", "buffer_utilization", "buffer_oldest_age", "storage_utilization", "storage_status", "transport_error_count", "collector_restart_count", "mapping_error_count", "validation_error_count", } ENDPOINT_RULES = { "contains": ({"plant", "asset", "device"}, {"asset", "device", "measurement_point"}), "feeds": ({"asset", "measurement_point"}, {"asset", "measurement_point"}), "connected_to": ({"asset", "measurement_point"}, {"asset", "measurement_point"}), "measures": ({"device", "measurement_point"}, {"asset", "measurement_point"}), "controls": ({"device"}, {"asset", "device"}), "powered_by": ({"asset", "device"}, {"asset", "measurement_point"}), } SECRET_KEYS = {"password", "secret", "token", "private_key", "api_key", "credential"} def parse_time(value): if not value: return None return datetime.fromisoformat(str(value).replace("Z", "+00:00")) def duplicates(values): seen = set() return {value for value in values if value in seen or seen.add(value)} def find_secret_keys(value, path=""): found = [] if isinstance(value, dict): for key, child in value.items(): child_path = f"{path}/{key}" if key.lower() in SECRET_KEYS: found.append(child_path) found.extend(find_secret_keys(child, child_path)) elif isinstance(value, list): for index, child in enumerate(value): found.extend(find_secret_keys(child, f"{path}/{index}")) return found def main() -> int: registry_path = Path(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_REGISTRY registry = yaml.safe_load(registry_path.read_text()) errors = [] if jsonschema is None: errors.append("jsonschema module not installed; schema check skipped") else: schema = json.loads(SCHEMA_PATH.read_text()) validator = jsonschema.Draft202012Validator(schema, format_checker=jsonschema.FormatChecker()) for err in sorted(validator.iter_errors(registry), key=lambda item: list(item.absolute_path)): location = "/".join(map(str, err.absolute_path)) errors.append(f"schema: {location}: {err.message}") organizations = registry.get("organizations", []) portfolios = registry.get("portfolios", []) plants = registry.get("plants", []) assets = registry.get("assets", []) devices = registry.get("devices", []) points = registry.get("measurement_points", []) topologies = registry.get("topologies", []) relationships = registry.get("relationships", []) org_by_id = {item.get("id"): item for item in organizations} portfolio_by_key = {(item.get("organization_id"), item.get("id")): item for item in portfolios} plant_by_id = {item.get("id"): item for item in plants} asset_by_key = {(item.get("plant_id"), item.get("id")): item for item in assets} device_by_key = {(item.get("plant_id"), item.get("id")): item for item in devices} point_by_key = {(item.get("plant_id"), item.get("id")): item for item in points} topology_by_key = {(item.get("plant_id"), item.get("id"), item.get("version")): item for item in topologies} for label, values in ( ("organization", [item.get("id") for item in organizations]), ("plant", [item.get("id") for item in plants]), ("portfolio", [(item.get("organization_id"), item.get("id")) for item in portfolios]), ("asset", [(item.get("plant_id"), item.get("id")) for item in assets]), ("device", [(item.get("plant_id"), item.get("id")) for item in devices]), ("measurement_point", [(item.get("plant_id"), item.get("id")) for item in points]), ("topology version", [(item.get("plant_id"), item.get("id"), item.get("version")) for item in topologies]), ("relationship", [(item.get("plant_id"), item.get("id")) for item in relationships]), ): for duplicate in duplicates(values): errors.append(f"duplicate {label} identity: {duplicate}") for portfolio in portfolios: if portfolio.get("organization_id") not in org_by_id: errors.append(f"portfolio '{portfolio.get('id')}' references unknown organization") for plant in plants: org_id = plant.get("organization_id") if org_id not in org_by_id: errors.append(f"plant '{plant.get('id')}' references unknown organization '{org_id}'") for portfolio_id in plant.get("portfolio_ids", []): if (org_id, portfolio_id) not in portfolio_by_key: errors.append(f"plant '{plant.get('id')}' references portfolio outside its organization: '{portfolio_id}'") for asset in assets: if asset.get("plant_id") not in plant_by_id: errors.append(f"asset '{asset.get('id')}' references unknown plant") for device in devices: plant_id = device.get("plant_id") if plant_id not in plant_by_id: errors.append(f"device '{device.get('id')}' references unknown plant") if (plant_id, device.get("asset_id")) not in asset_by_key: errors.append(f"device '{device.get('id')}' references unknown or cross-plant asset") for point in points: plant_id = point.get("plant_id") if plant_id not in plant_by_id: errors.append(f"measurement point '{point.get('id')}' references unknown plant") if (plant_id, point.get("device_id")) not in device_by_key: errors.append(f"measurement point '{point.get('id')}' references unknown or cross-plant device") parent = point.get("parent_measurement_point_id") if parent and (plant_id, parent) not in point_by_key: errors.append(f"measurement point '{point.get('id')}' references unknown parent") versions_by_topology = defaultdict(list) for topology in topologies: plant_id = topology.get("plant_id") if plant_id not in plant_by_id: errors.append(f"topology '{topology.get('id')}' references unknown plant") try: start, end = parse_time(topology.get("valid_from")), parse_time(topology.get("valid_to")) if end and end <= start: errors.append(f"topology '{topology.get('id')}' has an invalid validity interval") if topology.get("status") == "active": versions_by_topology[(plant_id, topology.get("id"))].append((start, end, topology.get("version"))) except ValueError: pass for key, intervals in versions_by_topology.items(): ordered_intervals = sorted(intervals, key=lambda item: (item[0], item[2])) for index, (start, end, version) in enumerate(ordered_intervals): for other_start, other_end, other_version in ordered_intervals[index + 1:]: if end is None or other_start < end: errors.append(f"active topology versions overlap for {key}: {version} and {other_version}") endpoints = { "plant": lambda plant_id, object_id: object_id == plant_id and object_id in plant_by_id, "asset": lambda plant_id, object_id: (plant_id, object_id) in asset_by_key, "device": lambda plant_id, object_id: (plant_id, object_id) in device_by_key, "measurement_point": lambda plant_id, object_id: (plant_id, object_id) in point_by_key, } connected_pairs = set() contains_graph = defaultdict(lambda: defaultdict(set)) for relationship in relationships: rel_id, plant_id = relationship.get("id"), relationship.get("plant_id") topology_key = (plant_id, relationship.get("topology_id"), relationship.get("topology_version")) if topology_key not in topology_by_key: errors.append(f"relationship '{rel_id}' references unknown topology version") source, target = relationship.get("from", {}), relationship.get("to", {}) for endpoint in (source, target): kind, object_id = endpoint.get("kind"), endpoint.get("id") if kind in endpoints and not endpoints[kind](plant_id, object_id): errors.append(f"relationship '{rel_id}' references unknown or cross-plant {kind} '{object_id}'") rel_type = relationship.get("type") if rel_type in ENDPOINT_RULES: allowed_from, allowed_to = ENDPOINT_RULES[rel_type] if source.get("kind") not in allowed_from or target.get("kind") not in allowed_to: errors.append(f"relationship '{rel_id}' has invalid endpoint kinds for '{rel_type}'") try: start, end = parse_time(relationship.get("valid_from")), parse_time(relationship.get("valid_to")) if end and end <= start: errors.append(f"relationship '{rel_id}' has an invalid validity interval") except ValueError: pass source_key = (source.get("kind"), source.get("id")) target_key = (target.get("kind"), target.get("id")) if rel_type == "connected_to": pair = (topology_key, tuple(sorted((source_key, target_key)))) if pair in connected_pairs: errors.append(f"relationship '{rel_id}' duplicates a symmetric connected_to pair") connected_pairs.add(pair) if rel_type == "contains": contains_graph[topology_key][source_key].add(target_key) for topology_key, graph in contains_graph.items(): visiting, visited = set(), set() def visit(node): if node in visiting: return True if node in visited: return False visiting.add(node) if any(visit(child) for child in graph.get(node, ())): return True visiting.remove(node) visited.add(node) return False if any(visit(node) for node in list(graph)): errors.append(f"contains relationship cycle in topology {topology_key}") for device in devices: org_id = plant_by_id.get(device.get("plant_id"), {}).get("organization_id", "") extension_prefix = f"x-{org_id}-" for capability in device.get("capabilities", []): if capability not in CATALOGUE_METRICS and not capability.startswith(extension_prefix): errors.append(f"device '{device.get('id')}' has unknown capability '{capability}'") if "edge_collector_observability" in device.get("profiles", []): if device.get("role") != "collector" or device.get("energy_form") != "operational": errors.append(f"device '{device.get('id')}' claims edge profile with invalid role or energy form") missing = EDGE_PROFILE_REQUIRED - set(device.get("capabilities", [])) if missing: errors.append(f"device '{device.get('id')}' is missing edge capabilities: {', '.join(sorted(missing))}") for secret_path in find_secret_keys(registry): errors.append(f"secret-like Registry key is forbidden: {secret_path}") if errors: print(f"INVALID: {registry_path}") for error in errors: print(f" - {error}") return 1 print( f"VALID: {registry_path} ({len(organizations)} organization(s), " f"{len(plants)} plant(s), {len(assets)} asset(s), {len(devices)} device(s), " f"{len(points)} measurement point(s), {len(topologies)} topology version(s), " f"{len(relationships)} relationship(s))" ) return 0 if __name__ == "__main__": sys.exit(main())