import streamlit as st import pandas as pd import os import json import folium import requests import yaml from streamlit_folium import st_folium from pathlib import Path from PIL import Image import base64 from io import BytesIO # --- UTILS --- @st.cache_data def get_image_base64(img_path): img = Image.open(img_path).convert("RGBA") w, h = img.size buffered = BytesIO() img.save(buffered, format="PNG") img_str = base64.b64encode(buffered.getvalue()).decode("ascii") return f"data:image/png;base64,{img_str}", w, h def _norm_mac_internal(s: str) -> str: """Standardizza il MAC in formato xx:xx:xx:xx:xx:xx minuscolo per il matching.""" s = (s or "").strip().replace("-", "").replace(":", "").replace(".", "").lower() if len(s) != 12: return s return ":".join([s[i:i+2] for i in range(0, 12, 2)]) def fetch_beacons_metadata(cfg): """Recupera l'elenco dei beacon dalle API e normalizza i MAC per il lookup.""" api_c = cfg.get("api", {}) token_url = api_c.get("token_url") beacons_url = api_c.get("get_beacons_url") try: secrets_path = os.environ.get("SECRETS_FILE") or "/config/secrets.yaml" with open(secrets_path, "r") as f: sec = yaml.safe_load(f).get("oidc", {}) # 1. Autenticazione OIDC payload = { "grant_type": "password", "client_id": api_c.get("client_id", "Fastapi"), "client_secret": sec.get("client_secret", ""), "username": sec.get("username", "core"), "password": sec.get("password", "") } resp_t = requests.post(token_url, data=payload, verify=False, timeout=5) if resp_t.status_code != 200: return {} token = resp_t.json().get("access_token") # 2. Download Beacon List e Normalizzazione headers = {"Authorization": f"Bearer {token}", "accept": "application/json"} resp_b = requests.get(beacons_url, headers=headers, verify=False, timeout=5) if resp_b.status_code == 200: # Dizionario {mac_normalizzato: nome_amichevole} return { _norm_mac_internal(it["mac"]): it.get("name", "N/A") for it in resp_b.json() if "mac" in it } except Exception as e: st.error(f"Errore recupero nomi beacon: {e}") return {} def show_inference_page(cfg): st.subheader("📡 Monitoraggio Beacon Real-Time") # --- CONFIGURAZIONE PERCORSI --- MAPS_DIR = Path(cfg['maps']['map_dir']) INFER_FILE = Path("/data/infer/infer.csv") # --- 1. SELEZIONE E STATO --- maps = sorted([f.replace(cfg['maps']['floor_prefix'], "").split('.')[0] for f in os.listdir(MAPS_DIR) if f.startswith(cfg['maps']['floor_prefix'])]) if not maps: st.warning("Nessuna mappa configurata.") return df_infer = pd.DataFrame() if INFER_FILE.exists(): df_infer = pd.read_csv(INFER_FILE, sep=";") c_piano, c_count, c_size = st.columns([3, 2, 2]) with c_piano: sub1, sub2 = st.columns([1, 1.2]) sub1.markdown("

Piano Visualizzato:

", unsafe_allow_html=True) floor_id = sub2.selectbox("", maps, key="inf_floor_v24", label_visibility="collapsed") df_active = df_infer[(df_infer['z'].astype(str) == str(floor_id)) & (df_infer['x'] != -1)] if not df_infer.empty else pd.DataFrame() with c_count: st.info(f"📡 Beacon Attivi: **{len(df_active)}**\n(Totali nel file: {len(df_infer)})") with c_size: m_size = st.slider("Dimensione Beacon:", 5, 20, 8, key="inf_msize_v24") meta_path = MAPS_DIR / f"{cfg['maps']['meta_prefix']}{floor_id}.json" if not meta_path.exists(): return with open(meta_path, "r") as f: meta = json.load(f) # --- 2. RENDERING MAPPA --- st.markdown("---") img_p = next((MAPS_DIR / f"{cfg['maps']['floor_prefix']}{floor_id}{e}" for e in ['.png','.jpg'] if (MAPS_DIR / f"{cfg['maps']['floor_prefix']}{floor_id}{e}").exists())) img_data, w, h = get_image_base64(img_p) bounds = [[0, 0], [h, w]] m = folium.Map(location=[h/2, w/2], crs="Simple", tiles=None, attribution_control=False) m.fit_bounds(bounds) m.options.update({"minZoom": -6, "maxZoom": 6, "zoomSnap": 0.25, "maxBounds": bounds, "maxBoundsViscosity": 1.0}) folium.raster_layers.ImageOverlay(image=img_data, bounds=bounds).add_to(m) # Recupero nomi per etichette mappa names_map = fetch_beacons_metadata(cfg) if meta["calibrated"] and meta["origin"] != [0,0]: # Origine (Punto di riferimento) folium.CircleMarker(location=[meta["origin"][1], meta["origin"][0]], radius=4, color="black", fill=True).add_to(m) for _, row in df_active.iterrows(): px_x = (row['x'] * meta["pixel_ratio"]) + meta["origin"][0] px_y = meta["origin"][1] - (row['y'] * meta["pixel_ratio"]) # --- UMANIZZAZIONE ETICHETTA MAPPA --- norm_mac = _norm_mac_internal(row['mac']) # Se disponibile usa il nome da API, altrimenti le ultime cifre del MAC label_text = names_map.get(norm_mac, str(row['mac'])[-5:]) # Pallino Beacon folium.CircleMarker( location=[px_y, px_x], radius=m_size, color="blue", fill=True, fill_color="cyan", fill_opacity=0.8, tooltip=f"Device: {label_text} | MAC: {row['mac']}" ).add_to(m) # Etichetta Nome (accanto al punto) folium.Marker( location=[px_y, px_x], icon=folium.DivIcon(html=f"""
{label_text}
""") ).add_to(m) st_folium(m, height=700, width=None, key=f"inf_map_v24_{floor_id}", use_container_width=True) # --- 3. TABELLA RIEPILOGO COMPLETA (Arricchita con Nome API) --- if not df_infer.empty: st.subheader("Dettaglio Dispositivi (Tutti i Piani)") df_display = df_infer.copy() df_display['mac_norm'] = df_display['mac'].apply(_norm_mac_internal) df_display['name'] = df_display['mac_norm'].map(names_map).fillna("Sconosciuto") # Selezione e riordino colonne finali cols_to_show = ['mac', 'name', 'z', 'x', 'y'] st.dataframe(df_display[cols_to_show], use_container_width=True)