Spaces:
Sleeping
Sleeping
| """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 | |