import re from typing import Mapping, Optional import pandas as pd def _normalize_profile_name(value: object) -> str: if value is None: return "" return str(value).strip().lower() def _normalize_site_key(value: object) -> str: if value is None or pd.isna(value): return "" s = str(value).strip() if not s: return "" if re.fullmatch(r"\d+(?:\.0+)?", s): return str(int(float(s))) return s.upper() def _iter_site_keys(*identifiers: object) -> list[str]: keys: list[str] = [] for identifier in identifiers: normalized = _normalize_site_key(identifier) if normalized and normalized not in keys: keys.append(normalized) if isinstance(identifier, str): raw = identifier.strip() if not raw: continue first_token = raw.split("_", 1)[0].strip() normalized_token = _normalize_site_key(first_token) if normalized_token and normalized_token not in keys: keys.append(normalized_token) return keys def parse_band_profile_definitions(profile_text: str) -> dict[str, str]: profiles: dict[str, str] = {} for line_no, raw_line in enumerate(profile_text.splitlines(), start=1): line = raw_line.strip() if not line or line.startswith("#"): continue if "=" not in line: raise ValueError( f"Invalid profile definition on line {line_no}: expected 'name = bands'." ) profile_name_raw, bands_raw = line.split("=", 1) profile_name = _normalize_profile_name(profile_name_raw) bands = str(bands_raw).strip() if not profile_name or not bands: raise ValueError( f"Invalid profile definition on line {line_no}: name and bands are required." ) profiles[profile_name] = bands return profiles def read_site_profile_mapping(mapping_file) -> dict[str, str]: if mapping_file is None: return {} if hasattr(mapping_file, "seek"): mapping_file.seek(0) df = pd.read_csv(mapping_file) normalized_columns = { str(column).strip().lower().replace(" ", "_"): column for column in df.columns } site_code_col = normalized_columns.get("site_code") profile_col = normalized_columns.get("profile") if site_code_col is None or profile_col is None: raise ValueError("Profile mapping CSV must contain columns 'site_code' and 'profile'.") mapping: dict[str, str] = {} for _, row in df.iterrows(): site_key = _normalize_site_key(row.get(site_code_col)) profile_name = _normalize_profile_name(row.get(profile_col)) if site_key and profile_name: mapping[site_key] = profile_name return mapping def resolve_site_bands( default_bands: str, profile_definitions: Optional[Mapping[str, str]], site_profile_mapping: Optional[Mapping[str, str]], *site_identifiers: object, ) -> str: bands = str(default_bands).strip() if not profile_definitions or not site_profile_mapping: return bands normalized_profiles = { _normalize_profile_name(profile_name): str(profile_bands).strip() for profile_name, profile_bands in profile_definitions.items() if _normalize_profile_name(profile_name) and str(profile_bands).strip() } if not normalized_profiles: return bands for site_key in _iter_site_keys(*site_identifiers): profile_name = site_profile_mapping.get(site_key) if not profile_name: continue profile_bands = normalized_profiles.get(_normalize_profile_name(profile_name)) if profile_bands: return profile_bands return bands