//! `python::som` — extraído do monólito na v0.10.0 //! (divisão SRP, crítica Qwen3.8 lote 3 nº 1; r-35 §2). use super::*; /// Treina um 4D-SOM e devolve telemetria completa (EQ/TE/massa/cobertura). #[pyfunction] pub(crate) fn treinar_som4d(py: Python, dados: Vec>, cfg: Option) -> PyResult { let dev = candle_core::Device::Cpu; let t = tensor_de_2d(dados, &dev).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("{e}")))?; let mut c = som4d::CfgSom4D::default(); if let Some(js) = &cfg { if let Ok(v) = serde_json::from_str::(js) { if let Some(o) = v.as_object() { for (k, val) in o { match k.as_str() { "nx" => c.nx = val.as_u64().unwrap_or(4) as usize, "ny" => c.ny = val.as_u64().unwrap_or(4) as usize, "nz" => c.nz = val.as_u64().unwrap_or(4) as usize, "nt" => c.nt = val.as_u64().unwrap_or(4) as usize, "epocas" => c.epocas = val.as_u64().unwrap_or(50) as usize, "sigma0" => c.sigma0 = val.as_f64().unwrap_or(2.0), "sigma1" => c.sigma1 = val.as_f64().unwrap_or(0.5), "eta0" => c.eta0 = val.as_f64().unwrap_or(0.5), "eta1" => c.eta1 = val.as_f64().unwrap_or(0.05), "semente" => c.semente = val.as_u64().unwrap_or(42), "espalhada" => { c.inicializacao = if val.as_bool().unwrap_or(false) { som4d::Inicializacao4D::Espalhada } else { som4d::Inicializacao4D::Dados } } _ => {} } } } } } let som = som4d::treinar_som4d(&t, &c, &dev) .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("{e}")))?; let d = serde_json::json!({ "versao": som4d::versao(), "celulas": som.n_celulas(), "grade": [som.nx, som.ny, som.nz, som.nt], "eq_inicial": som.eq_inicial, "eq_final": som.eq_final, "te_final": som.te_final, "historico_eq_te": som.historia, "massa_soma": som.massa.iter().sum::(), }); Ok(pythonize::pythonize(py, &d)?.into()) } /// Treina um GNG e devolve telemetria (vértices, arestas, componentes, EQ). #[pyfunction] pub(crate) fn treinar_gng(py: Python, dados: Vec>, cfg: Option) -> PyResult { let dev = candle_core::Device::Cpu; let t = tensor_de_2d(dados, &dev).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("{e}")))?; let mut c = som4d::CfgGng::default(); if let Some(js) = &cfg { if let Ok(v) = serde_json::from_str::(js) { if let Some(o) = v.as_object() { for (k, val) in o { match k.as_str() { "passos" => c.passos = val.as_u64().unwrap_or(4000) as usize, "max_vertices" => c.max_vertices = val.as_u64().unwrap_or(60) as usize, "lambda" => c.lambda = val.as_u64().unwrap_or(50) as usize, "eps_b" => c.eps_b = val.as_f64().unwrap_or(0.05), "eps_n" => c.eps_n = val.as_f64().unwrap_or(0.006), "idade_max" => c.idade_max = val.as_u64().unwrap_or(50) as u32, "semente" => c.semente = val.as_u64().unwrap_or(42), _ => {} } } } } } let g = som4d::treinar_gng(&t, &c) .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("{e}")))?; let d = serde_json::json!({ "versao": som4d::versao(), "vertices": g.vertices.len(), "arestas": g.arestas.len(), "componentes_conexas": g.n_componentes_conexas(), "insercoes": g.insercoes, "eq_medio": g.eq_medio, }); Ok(pythonize::pythonize(py, &d)?.into()) } /// Treina um H-SOM e devolve a decomposição hierárquica (Teo H-1). #[pyfunction] pub(crate) fn treinar_hsom(py: Python, dados: Vec>, cfg: Option) -> PyResult { let dev = candle_core::Device::Cpu; let t = tensor_de_2d(dados, &dev).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("{e}")))?; let mut c = som4d::CfgHSom::default(); if let Some(js) = &cfg { if let Ok(v) = serde_json::from_str::(js) { if let Some(o) = v.as_object() { if let Some(x) = o.get("celulas_finas") { c.celulas_finas = x.as_u64().unwrap_or(9) as usize; } if let Some(x) = o.get("epocas_finas") { c.epocas_finas = x.as_u64().unwrap_or(20) as usize; } } } } let h = som4d::treinar_hsom(&t, &c, &dev) .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("{e}")))?; let cota = h.eq_grosso + 2.0 * h.r_fino + 2.0 * (h.eq_grosso * h.r_fino).sqrt(); let d = serde_json::json!({ "versao": som4d::versao(), "eq_hierarquico": h.eq_hierarquico, "eq_grosso": h.eq_grosso, "r_fino": h.r_fino, "cota_teo_h1": cota, "cota_respeitada": h.eq_hierarquico <= cota + 1e-9, }); Ok(pythonize::pythonize(py, &d)?.into()) } /// Demo T-SOM: streaming com deriva controlada — rastreio (Teo T-1). #[pyfunction] pub(crate) fn demo_tsom(py: Python, dados_iniciais: Vec>, janela: Vec>, passos: usize, gamma: f64) -> PyResult { let dev = candle_core::Device::Cpu; let t0 = tensor_de_2d(dados_iniciais, &dev).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("{e}")))?; let cfg = som4d::CfgTSom { cfg_grade: som4d::CfgSom4D { nx: 3, ny: 3, nz: 2, nt: 2, epocas: 15, ..Default::default() }, gamma, janela: janela.len(), }; let mut ts = som4d::TSom4D::novo(&t0, &cfg, &dev) .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("{e}")))?; let mut historico = Vec::new(); for _ in 0..passos { let eq = ts.passo(&janela); historico.push(eq); } let d = serde_json::json!({ "versao": som4d::versao(), "gamma": gamma, "eq_temporal_historico": historico, "t_passos": ts.t, }); Ok(pythonize::pythonize(py, &d)?.into()) } /// Demo/gate do auto-ajuste SOM com tetos (S-AB/S-SL). #[pyfunction] pub(crate) fn demo_autoajuste_som(py: Python) -> PyResult { let mut a = controlador::AutoAjusteSom::novo(0.5, 1.0); let (al_qe, _) = a.passo(1.0, 0.0, 0.0); let (al_cross, sg_cross) = a.passo(0.0, 0.0, 1.0); let (al_max, sg_max) = a.passo(1.0, 1.0, 1.0); let mut parou_5chunks = false; for _ in 0..5 { parou_5chunks = a.avaliar_parada(0.95, 0.1, 0.1, false, true, false, false); } let d_ = serde_json::json!({ "versao": "som-autoadjust-0.7.0", "alfa_qe_alta": al_qe, "alfa_crossing": al_cross, "sigma_crossing": sg_cross, "alfa_max": al_max, "sigma_max": sg_max, "parou_5chunks": parou_5chunks, "motivo": a.motivo_parada, "gates": { "tetos_ok": al_max <= 1.0 + 1e-9 && sg_max <= 2.0 + 1e-9, "piso_ok": al_cross >= 0.5 - 1e-9, "qe_sobe": al_qe > 0.5, "stop_dispara": parou_5chunks, }, }); Ok(pythonize::pythonize(py, &d_)?.into()) }