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