"""Use the imported model snapshot; local names alone do not establish identity.""" import hashlib import json from pathlib import Path import numpy as np import pandas as pd from . import get_model, list_available_models from .model_data import ModelData from .anos import detectar_ano_base, periodo_variavel_ano def context_key(row): terrain = row.get('evolutivo', {}).get('terreno', {}) context = terrain.get('contextoModelo') or {} return row.get('modelo', ''), hashlib.sha256(json.dumps(context, sort_keys=True, ensure_ascii=False).encode()).hexdigest()[:12] def matching_local(row): name = row.get('modelo') context = row.get('evolutivo', {}).get('terreno', {}).get('contextoModelo') or {} if name not in list_available_models(): return None, 'Modelo indisponível no catálogo.' local = get_model(name) if not context: return None, 'O arquivo não preserva o contexto necessário para confirmar a versão do modelo do catálogo.' prediction = context.get('prediction') or {} names, params = prediction.get('names') or [], prediction.get('params') or [] sm = local.modelos_sm if local else None try: local_names = list(map(str, sm.model.exog_names)) same = names == local_names and len(params) == len(sm.params) and np.allclose(params, sm.params, rtol=1e-12, atol=1e-12) # Equal parameters are necessary, but not sufficient: verify fitted data too. x = context.get('transformacoes', {}).get('X') y = context.get('transformacoes', {}).get('y') portable_x = pd.DataFrame(x).reindex(columns=names, fill_value=1).to_numpy(dtype=float) portable_y = np.asarray(y, dtype=float).reshape(-1) same = same and portable_x.shape == sm.model.exog.shape and portable_y.shape == sm.model.endog.shape same = same and np.allclose(portable_x, sm.model.exog, rtol=1e-12, atol=1e-12) and np.allclose(portable_y, sm.model.endog, rtol=1e-12, atol=1e-12) covariance = np.asarray(prediction.get('covariance'),dtype=float) same = same and covariance.shape == np.asarray(sm.cov_params()).shape and np.allclose(covariance,sm.cov_params(),rtol=1e-12,atol=1e-12) same = same and context.get('transformacoes',{}).get('info') == local.formatted_top_transformation_info if same: return local, '' except (AttributeError, TypeError, ValueError, KeyError): pass return None, 'A correspondência da versão do catálogo não foi confirmada com os coeficientes e a amostra salvos; os anexos desse modelo não serão usados.' def snapshot_model(row): context = row.get('evolutivo', {}).get('terreno', {}).get('contextoModelo') or {} prediction = context.get('prediction') or {} names, params = prediction.get('names') or [], prediction.get('params') or [] if not names or len(names) != len(params): return None raw = context.get('dados', {}).get('df') or {} try: df = pd.DataFrame(**raw) if isinstance(raw, dict) and 'columns' in raw else pd.DataFrame(raw) except (ValueError, TypeError): df = pd.DataFrame() transforms = context.get('transformacoes') or {} info = transforms.get('info') or [] mapping = dict(item.split(': ', 1) for item in info if isinstance(item, str) and ': ' in item) # Transcribe the fitted expression; do not fit or evaluate a new model. terms = [] for name, param in zip(names, params): coefficient = format(float(param), '.12g') if name == 'const': terms.append(coefficient) else: transformed = mapping.get(name, '(x)').replace('x', f'[{name}]') terms.append(f'({coefficient} × {transformed})') dependent = next((item.split(': ', 1) for item in info if isinstance(item, str) and ': ' in item and item.split(': ', 1)[0] not in names), ['Y', 'y']) lhs = dependent[1].replace('y', dependent[0]) evidence = context.get('normativeEvidence') or {} annual_names = [str(name) for name in names if periodo_variavel_ano(str(name))] reference = context.get('referencia_temporal') base_year = detectar_ano_base(df, annual_names, reference) if base_year is None and reference is None: # Legacy EVO contexts omitted the declaration. A name alone is not # enough to borrow it: matching_local verifies the fitted model/data. local, _ = matching_local(row) if local is not None: base_year = local.ano_base return ModelData( nome=row.get('modelo') or context.get('name') or 'Modelo importado', path=Path('contexto_importado'), xy_preview=df, tabelas_coef=pd.DataFrame({'coef': params}, index=names), modelos_resumos={'n': evidence.get('n', len(df)), 'k': evidence.get('k', len(names)-1), 'r2': evidence.get('r2'), 'equacao': f'{lhs} = ' + ' + '.join(terms)}, formatted_top_transformation_info=info, variaveis_dicotomicas=transforms.get('dicotomicas') or [], periodo_dados_mercado=context.get('periodo_dados_mercado') or {}, ano_base=base_year, )