""" GOID Generator - Global Object ID for Infrastructure Object Model Format: [DOM]-[SYS]-[SUB]-[OBJ]-[SEQ] Example: BYG-FAC-WIN-GLA-0001 """ import hashlib import re import json import os from typing import Optional, Dict, List, Tuple from datetime import datetime class GOIDGenerator: """Generates and validates Global Object IDs for IOM""" def __init__(self, taxonomy_path: Optional[str] = None): self.sequence_counters: Dict[str, int] = {} self.taxonomy = self._load_taxonomy(taxonomy_path) def _load_taxonomy(self, path: Optional[str]) -> Dict: """Load taxonomy from JSON file""" if path is None: path = os.path.join(os.path.dirname(__file__), 'iom_taxonomy.json') with open(path, 'r', encoding='utf-8') as f: return json.load(f) def generate( self, domain: str, system: str, subsystem: str, obj_type: str, location_hash: Optional[str] = None ) -> str: """ Generate a unique GOID Args: domain: 3-letter domain code (e.g., 'BYG') system: 3-letter system code (e.g., 'FAC') subsystem: 3-letter subsystem code (e.g., 'WIN') obj_type: 3-letter object type code (e.g., 'GLA') location_hash: Optional location hash for extra uniqueness Returns: GOID string like 'BYG-FAC-WIN-GLA-0001' """ # Validate codes exist in taxonomy self._validate_codes(domain, system, subsystem, obj_type) # Build prefix prefix = f"{domain}-{system}-{subsystem}-{obj_type}" # Get next sequence number if prefix not in self.sequence_counters: self.sequence_counters[prefix] = 0 self.sequence_counters[prefix] += 1 # Format: DOM-SYS-SUB-OBJ-SEQ goid = f"{prefix}-{self.sequence_counters[prefix]:04d}" # Add location hash if provided if location_hash: short_hash = hashlib.md5(location_hash.encode()).hexdigest()[:4].upper() goid = f"{goid}-{short_hash}" return goid def validate(self, goid: str) -> Tuple[bool, Optional[str]]: """ Validate GOID format Returns: Tuple of (is_valid, error_message) """ if not goid: return False, "GOID is empty" parts = goid.split('-') # Minimum 5 parts: DOM-SYS-SUB-OBJ-SEQ if len(parts) < 5: return False, f"GOID must have at least 5 parts, got {len(parts)}" # Validate domain, system, subsystem, object type (3 letters each) for i, part in enumerate(parts[:4]): if len(part) != 3: return False, f"Part {i+1} must be 3 letters, got '{part}' ({len(part)} chars)" if not part.isalpha(): return False, f"Part {i+1} must be letters only, got '{part}'" if not part.isupper(): return False, f"Part {i+1} must be uppercase, got '{part}'" # Validate sequence number try: seq = int(parts[4]) if seq < 1: return False, f"Sequence must be >= 1, got {seq}" except ValueError: return False, f"Part 5 must be a number, got '{parts[4]}'" # Validate optional location hash (if present) if len(parts) == 6: if len(parts[5]) != 4: return False, f"Location hash must be 4 chars, got '{parts[5]}'" if not all(c.isalnum() for c in parts[5]): return False, f"Location hash must be alphanumeric, got '{parts[5]}'" if len(parts) > 6: return False, f"GOID has too many parts: {len(parts)}" return True, None def parse(self, goid: str) -> Dict: """ Parse a GOID into its components Returns: Dict with domain, system, subsystem, obj_type, sequence, location_hash """ is_valid, error = self.validate(goid) if not is_valid: raise ValueError(f"Invalid GOID: {error}") parts = goid.split('-') result = { 'domain': parts[0], 'system': parts[1], 'subsystem': parts[2], 'obj_type': parts[3], 'sequence': int(parts[4]), 'location_hash': parts[5] if len(parts) == 6 else None, 'full_goid': goid } # Add human-readable names from taxonomy result['domain_name'] = self._get_domain_name(parts[0]) result['system_name'] = self._get_system_name(parts[0], parts[1]) result['subsystem_name'] = self._get_subsystem_name(parts[0], parts[1], parts[2]) return result def _validate_codes(self, domain: str, system: str, subsystem: str, obj_type: str): """Validate that codes exist in taxonomy""" domains = self.taxonomy.get('domains', {}) if domain not in domains: raise ValueError(f"Unknown domain: '{domain}'. Available: {list(domains.keys())}") systems = domains[domain].get('systems', {}) if system not in systems: raise ValueError(f"Unknown system '{system}' in domain '{domain}'. Available: {list(systems.keys())}") objects = systems[system].get('objects', {}) if subsystem not in objects: raise ValueError(f"Unknown subsystem '{subsystem}' in system '{system}'. Available: {list(objects.keys())}") # obj_type should be a component or the subsystem itself obj_data = objects[subsystem] valid_types = [subsystem] + obj_data.get('components', []) if obj_type not in valid_types: raise ValueError(f"Unknown object type '{obj_type}' in subsystem '{subsystem}'. Available: {valid_types}") def _get_domain_name(self, domain: str) -> str: """Get human-readable domain name""" domains = self.taxonomy.get('domains', {}) return domains.get(domain, {}).get('name', domain) def _get_system_name(self, domain: str, system: str) -> str: """Get human-readable system name""" domains = self.taxonomy.get('domains', {}) systems = domains.get(domain, {}).get('systems', {}) return systems.get(system, {}).get('name', system) def _get_subsystem_name(self, domain: str, system: str, subsystem: str) -> str: """Get human-readable subsystem name""" domains = self.taxonomy.get('domains', {}) systems = domains.get(domain, {}).get('systems', {}) objects = systems.get(system, {}).get('objects', {}) return objects.get(subsystem, {}).get('name', subsystem) def list_domains(self) -> List[Dict]: """List all available domains""" domains = self.taxonomy.get('domains', {}) return [ {'code': code, 'name': data['name'], 'name_en': data.get('name_en', '')} for code, data in domains.items() ] def list_systems(self, domain: str) -> List[Dict]: """List all systems in a domain""" domains = self.taxonomy.get('domains', {}) systems = domains.get(domain, {}).get('systems', {}) return [ {'code': code, 'name': data['name'], 'name_en': data.get('name_en', '')} for code, data in systems.items() ] def list_objects(self, domain: str, system: str) -> List[Dict]: """List all objects in a system""" domains = self.taxonomy.get('domains', {}) systems = domains.get(domain, {}).get('systems', {}) objects = systems.get(system, {}).get('objects', {}) return [ { 'code': code, 'name': data['name'], 'name_en': data.get('name_en', ''), 'components': data.get('components', []) } for code, data in objects.items() ] # Singleton instance goid_generator = GOIDGenerator() def generate_goid( domain: str, system: str, subsystem: str, obj_type: str, location_hash: Optional[str] = None ) -> str: """Convenience function using singleton""" return goid_generator.generate(domain, system, subsystem, obj_type, location_hash) def validate_goid(goid: str) -> Tuple[bool, Optional[str]]: """Convenience function using singleton""" return goid_generator.validate(goid) def parse_goid(goid: str) -> Dict: """Convenience function using singleton""" return goid_generator.parse(goid) if __name__ == '__main__': # Example usage gen = GOIDGenerator() # Generate some GOIDs print("=== Generating GOIDs ===") goid1 = gen.generate('BYG', 'FAC', 'WIN', 'GLA') print(f"Window glass: {goid1}") goid2 = gen.generate('BYG', 'FAC', 'WIN', 'GLA') print(f"Window glass 2: {goid2}") goid3 = gen.generate('ENE', 'EVC', 'CHA', 'LED', location_hash='59.3293,18.0686') print(f"EV charger with location: {goid3}") # Validate print("\n=== Validation ===") for g in [goid1, goid2, goid3, 'INVALID', 'BYG-FAC-WIN-GLA']: is_valid, error = gen.validate(g) print(f"{g}: {'✓' if is_valid else '✗'} {error or ''}") # Parse print("\n=== Parsing ===") parsed = gen.parse(goid1) print(f"Parsed {goid1}:") for key, value in parsed.items(): print(f" {key}: {value}") # List domains print("\n=== Available Domains ===") for domain in gen.list_domains(): print(f" {domain['code']}: {domain['name']} ({domain['name_en']})")