""" Placeholder/Metric Registry Framework Central registry for all placeholders/metrics ensuring consistent metadata across: - Backend prompt resolution (Layer 2a) - GUI selection lists - Extended export - Validation - Chart assignment (Layer 2b) Version: 1.0 (Part A - Nutrition Basis Metrics) """ from dataclasses import dataclass, field, asdict from typing import Callable, Dict, List, Optional, Any from enum import Enum class EvidenceType(str, Enum): """Evidence type for metadata fields.""" CODE_DERIVED = "code-derived" DRAFT_DERIVED = "draft-derived" MIXED = "mixed" UNRESOLVED = "unresolved" TO_VERIFY = "to_verify" class OutputType(str, Enum): """Placeholder output types.""" NUMERIC = "numeric" STRING = "string" BOOLEAN = "boolean" JSON = "json" LIST = "list" TEXT_SUMMARY = "text_summary" class PlaceholderType(str, Enum): """Placeholder semantic types.""" ATOMIC = "atomic" RAW_DATA = "raw_data" INTERPRETED = "interpreted" SCORE = "score" META = "meta" @dataclass class MissingValuePolicy: """Structured missing value handling.""" available: bool value_raw: Optional[Any] missing_reason: str # no_data, insufficient_data, resolver_error, calculation_error, not_applicable legacy_display: str @dataclass class PlaceholderMetadata: """ Complete metadata for a placeholder/metric. All fields track their evidence type to maintain transparency about what is code-derived vs. draft-derived. """ # Core identification key: str category: str description: str # Technical (typically code-derived) resolver_module: str resolver_function: str data_layer_module: Optional[str] = None data_layer_function: Optional[str] = None source_tables: List[str] = field(default_factory=list) # Semantic (typically draft-derived or mixed) semantic_contract: str = "" business_meaning: str = "" unit: str = "" time_window: str = "" output_type: OutputType = OutputType.STRING placeholder_type: PlaceholderType = PlaceholderType.INTERPRETED format_hint: str = "" example_output: str = "" # Quality (mixed sources) minimum_data_requirements: Optional[str] = None quality_filter_policy: Optional[str] = None confidence_logic: Optional[str] = None missing_value_policy: Optional[MissingValuePolicy] = None known_limitations: Optional[str] = None # Architecture (code-derived) layer_1_decision: Optional[str] = None layer_2a_decision: Optional[str] = None layer_2b_reuse_possible: Optional[bool] = None architecture_alignment: Optional[str] = None issue_53_alignment: Optional[str] = None # Evidence tracking evidence: Dict[str, EvidenceType] = field(default_factory=dict) # Runtime resolver (not serialized to export) _resolver_func: Optional[Callable] = field(default=None, repr=False, compare=False) def to_dict(self, include_resolver: bool = False) -> Dict: """Convert to dictionary for export.""" data = asdict(self) # Remove private fields if not include_resolver: data.pop('_resolver_func', None) # Convert enums to strings data['output_type'] = self.output_type.value data['placeholder_type'] = self.placeholder_type.value # Convert evidence dict data['evidence'] = {k: v.value for k, v in self.evidence.items()} # Convert missing_value_policy if self.missing_value_policy: data['missing_value_policy'] = asdict(self.missing_value_policy) return data def get_evidence(self, field_name: str) -> Optional[EvidenceType]: """Get evidence type for a field.""" return self.evidence.get(field_name) def set_evidence(self, field_name: str, evidence_type: EvidenceType): """Set evidence type for a field.""" self.evidence[field_name] = evidence_type def validate(self) -> List[str]: """Validate metadata completeness.""" issues = [] if not self.key: issues.append("Missing key") if not self.category: issues.append("Missing category") if not self.description: issues.append("Missing description") if not self.resolver_module: issues.append("Missing resolver_module") if not self.resolver_function: issues.append("Missing resolver_function") if not self.semantic_contract: issues.append("Missing semantic_contract") if not self.unit: issues.append("Missing unit") if not self.time_window: issues.append("Missing time_window") return issues class PlaceholderRegistry: """ Central registry for all placeholders/metrics. Ensures single source of truth for metadata across all consumers. """ def __init__(self): self._registry: Dict[str, PlaceholderMetadata] = {} def register( self, metadata: PlaceholderMetadata, resolver_func: Optional[Callable] = None ): """ Register a placeholder with complete metadata. Args: metadata: Complete placeholder metadata resolver_func: Optional resolver function (for runtime resolution) """ if metadata.key in self._registry: raise ValueError(f"Placeholder {metadata.key} already registered") if resolver_func: metadata._resolver_func = resolver_func self._registry[metadata.key] = metadata def get(self, key: str) -> Optional[PlaceholderMetadata]: """Get metadata for a placeholder.""" return self._registry.get(key) def get_all(self) -> Dict[str, PlaceholderMetadata]: """Get all registered placeholders.""" return self._registry.copy() def get_by_category(self, category: str) -> List[PlaceholderMetadata]: """Get placeholders by category (for GUI selection lists).""" return [ m for m in self._registry.values() if m.category == category ] def get_all_for_export(self) -> List[Dict]: """Get all metadata for extended export.""" return [m.to_dict() for m in self._registry.values()] def get_by_evidence_type(self, evidence_type: EvidenceType) -> Dict[str, List[str]]: """ Get fields by evidence type (for quality assurance). Returns: Dict mapping placeholder_key to list of field_names with that evidence type """ result = {} for key, metadata in self._registry.items(): fields = [ field_name for field_name, ev_type in metadata.evidence.items() if ev_type == evidence_type ] if fields: result[key] = fields return result def validate_all(self) -> Dict[str, List[str]]: """ Validate all registered placeholders. Returns: Dict mapping placeholder_key to list of validation issues """ issues = {} for key, metadata in self._registry.items(): validation_issues = metadata.validate() if validation_issues: issues[key] = validation_issues return issues def resolve(self, key: str, profile_id: str) -> str: """ Resolve a placeholder value for a profile. Args: key: Placeholder key profile_id: User profile ID Returns: Resolved value as string """ metadata = self.get(key) if not metadata: raise ValueError(f"Placeholder {key} not registered") if not metadata._resolver_func: raise ValueError(f"Placeholder {key} has no resolver function") return metadata._resolver_func(profile_id) # Global registry instance _global_registry = PlaceholderRegistry() def get_registry() -> PlaceholderRegistry: """Get the global placeholder registry.""" return _global_registry def register_placeholder( metadata: PlaceholderMetadata, resolver_func: Optional[Callable] = None ): """ Register a placeholder in the global registry. Args: metadata: Complete placeholder metadata resolver_func: Optional resolver function """ _global_registry.register(metadata, resolver_func)