import streamlit as st
import pandas as pd
from collections import defaultdict
import datetime
import time
st.set_page_config(layout="wide")
st.title("🧬 ABS Decay Analyzer DOC v1.3 ☕")
st.caption("Ultima versione: 2025-04-17")
# --- Funzioni di parsing ---
def parse_group_input(input_str):
return [x.strip().zfill(2) for x in input_str.replace('\n', '.').replace(',', '.').replace(' ', '.').split('.') if x.strip()]
def load_extractions(file):
lines = file.getvalue().decode("utf-8").strip().split("\n")
estrazioni = [line.strip().split('.') for line in lines if line.strip()]
return estrazioni
# --- UI ---
uploaded_file = st.file_uploader("Carica archivio estrazioni (.txt)", type=["txt"])
gruppo_base_input = st.text_area("Inserisci gruppo ABS iniziale (formato 01.02.03...)", height=100)
classe_finale = st.slider("Classe finale da monitorare (ridotta dalla ABS)", 1, 90, 9)
sorte = st.selectbox("Sorte da verificare", ["1", "2", "3", "4", "5"], index=1)
ruota_label = st.text_input("Etichetta ruota (es. GE)", value="GE")
num_ultime = st.slider("Quante ultime estrazioni vuoi analizzare?", min_value=50, max_value=10631, value=10631, step=50)
if uploaded_file and gruppo_base_input:
gruppo_base = parse_group_input(gruppo_base_input)
estrazioni = load_extractions(uploaded_file)
estrazioni = estrazioni[:num_ultime]
from itertools import combinations
max_combinazioni = 100000
sviluppo = []
counter = 0
st.info("Generazione formazioni in corso...")
for c in combinations(gruppo_base, classe_finale):
sviluppo.append(c)
counter += 1
if counter >= max_combinazioni:
st.warning(f"Limite massimo di {max_combinazioni} combinazioni raggiunto. Riduci la classe o il gruppo base per elaborare tutto.")
break
storico = defaultdict(lambda: {"max_colpo": 0, "frequenze": 0})
ritardi_correnti = defaultdict(int)
casi_attuali = []
progress = st.progress(0, text="Analisi estrazioni in corso...")
for i in range(len(estrazioni) - 1, -1, -1):
estrazione = estrazioni[i]
for formazione in sviluppo:
key = '.'.join(sorted(formazione))
ritardi_correnti[key] += 1
match = any(set(comb).issubset(set(estrazione)) for comb in combinations(formazione, int(sorte)))
if match:
storico[key]["frequenze"] += 1
storico[key]["max_colpo"] = max(storico[key]["max_colpo"], ritardi_correnti[key])
ritardi_correnti[key] = 0
if i % 10 == 0 or i == 0:
progress.progress((len(estrazioni) - i) / len(estrazioni), text=f"Analisi estrazione {len(estrazioni) - i} di {len(estrazioni)}")
progress.empty()
for key in sviluppo:
formazione = '.'.join(sorted(key))
if storico[formazione]["frequenze"] > 0:
rit_attuale = ritardi_correnti[formazione]
rit_max = storico[formazione]["max_colpo"]
delta = rit_max - rit_attuale
stato = "🟢" if delta > 3 else "🟡" if delta >= 0 else "🔴"
if delta <= 3:
casi_attuali.append({
"Numero" if classe_finale == 1 else "Formazione": formazione,
"Ritardo massimo storico": rit_max,
"Ritardo attuale": rit_attuale,
"Δ (Max - Attuale)": delta,
"Stato": stato
})
if casi_attuali:
df = pd.DataFrame(casi_attuali).sort_values("Δ (Max - Attuale)")
st.subheader(f"Formazioni attualmente in tensione storica su {ruota_label.upper()}")
st.dataframe(df.reset_index(drop=True), use_container_width=True)
st.markdown(f"Totale formazioni attive: **{len(df)}**")
if st.button("Scarica CSV"):
st.download_button("Download CSV", data=df.to_csv(index=False), file_name="abs_decay_report.csv")
else:
st.info("Nessuna formazione in tensione rilevata con i parametri attuali.")