"""Identificação das variáveis temporais e da referência registrada no modelo.""" import re from typing import Optional import pandas as pd def periodo_variavel_ano(nome: str) -> Optional[tuple[tuple[int, ...], bool]]: """Retorna os anos da dicotômica e se ela significa 'a partir de'.""" nome = str(nome) ano = r"((?:19|20)\d{2}|2100)" individual = re.fullmatch(rf"(?:a|ANO_?)?{ano}", nome) if individual: return (int(individual[1]),), False a_partir = re.fullmatch(rf"AP{ano}", nome) if a_partir: return (int(a_partir[1]),), True grupo = re.fullmatch(r"ANOS_((?:(?:19|20)\d{2}|2100)(?:_(?:(?:19|20)\d{2}|2100))+)", nome) if grupo: return tuple(int(valor) for valor in grupo[1].split("_")), False abreviado = re.fullmatch(rf"{ano}/(\d{{2}})", nome) if abreviado: primeiro = int(abreviado[1]) return (primeiro, primeiro // 100 * 100 + int(abreviado[2])), False return None def ano_base_explicito(referencia_temporal: dict, colunas_ano: list[str]) -> Optional[int]: """Read the declared reference only when it belongs to these annual variables.""" if not isinstance(referencia_temporal, dict) or referencia_temporal.get('tipo') != 'ano': return None bases = referencia_temporal.get('anos_base') variaveis = referencia_temporal.get('variaveis_ano') if (not isinstance(bases, list) or len(bases) != 1 or not isinstance(variaveis, list) or not colunas_ano or set(map(str, variaveis)) != set(map(str, colunas_ano))): return None if not re.fullmatch(r'(?:19|20)\d{2}|2100', str(bases[0])): return None base = int(bases[0]) for nome in colunas_ano: periodo = periodo_variavel_ano(str(nome)) if periodo is None: return None anos, a_partir = periodo if (a_partir and base >= anos[0]) or (not a_partir and base in anos): return None return base def detectar_ano_base(df: pd.DataFrame, colunas_ano: list[str], referencia_temporal=None) -> Optional[int]: """Confere o ano original dos registros em que todas as dicotômicas são zero. Uma referência ausente, ambígua ou incompatível com as dicotômicas não é substituída por um ano presumido. """ # New DAI files declare the reference independently of a source ANO column. # Keep inference from original data for older files only. if referencia_temporal is not None: return ano_base_explicito(referencia_temporal, colunas_ano) if not isinstance(df, pd.DataFrame) or df.empty or not colunas_ano: return None periodos = {coluna: periodo_variavel_ano(coluna) for coluna in colunas_ano} if any(periodo is None for periodo in periodos.values()): return None # Alguns arquivos antigos usam inteiros como nomes das colunas de ano. nomes = {str(coluna): coluna for coluna in df.columns} if any(coluna not in nomes for coluna in colunas_ano): return None anuais = df[[nomes[coluna] for coluna in colunas_ano]].apply(pd.to_numeric, errors="coerce") if anuais.isna().any().any() or not anuais.isin([0, 1]).all().all(): return None if anuais.sum(axis=1).gt(1).any(): return None registros_base = anuais.eq(0).all(axis=1) if not registros_base.any(): return None candidatos = [] for coluna in df.columns: if re.sub(r"[\s_-]", "", str(coluna)).casefold() not in ("ano", "anobase"): continue anos = pd.to_numeric(df[coluna], errors="coerce") if anos.isna().any() or not anos.between(1900, 2100).all() or not anos.mod(1).eq(0).all(): continue consistente = True for nome, (periodo, a_partir) in periodos.items(): esperado = anos.ge(periodo[0]) if a_partir else anos.isin(periodo) if not anuais[nomes[nome]].eq(esperado.astype(int)).all(): consistente = False break referencias = anos[registros_base].unique() if consistente and len(referencias) == 1: candidatos.append(int(referencias[0])) return candidatos[0] if candidatos and len(set(candidatos)) == 1 else None