import streamlit as st from const import BLURB from streamlit.runtime.media_file_storage import MediaFileStorageError import pandas as pd from pymongo.errors import PyMongoError import datetime from preload import mongo_client #### Logging def store_data(data, mongo_client): try: mongo_client["poverta_educativa"]["logging"].insert_one(data) except PyMongoError as e: st.toast("Could not connect to our database!") def save_report(session_state, modal=None, error=None): if error is not None: _save_report(session_state, modal=modal, error=error) elif modal is not None and session_state["logging_level"] == "modal": _save_report(session_state, modal=modal, error=error) elif session_state["logging_level"] == "advanced": _save_report(session_state, modal=modal, error=error) def _save_report(session_state, modal=None, error=None): data = { "UniqueID": session_state["id"], "Timestamp": str(datetime.datetime.now()), "Start": False, } if session_state["view"]: data["view"] = { "map_name": session_state["view"].get_map_name(), "hierarchy": session_state["view"].hierarchy, "hierarchy_names": session_state["view"].hierarchy_names, } data["filters"] = { "deprivation": session_state["filter_deprivation"], "objective": session_state["filter_objective"], "activities": session_state["filter_activities"], "target": session_state["filter_target"], } data["modal"] = modal data["error"] = error store_data(data, mongo_client) #### SIDEBAR # This part of the code deals with the sidebar def display_sidebar(): with st.sidebar: try: st.image("./graphics/Logo_CdFdOBdPdL.png") except MediaFileStorageError: st.write("Alternative Text: Logo") st.header( "Mappatura dei progetti di contrasto alla Povertà Educativa attivi in Piemonte, Liguria e Valle d'Aosta" ) for item in BLURB: st.write(item) try: st.image("./graphics/Logo_ASVAPP.png", width=60) except MediaFileStorageError: st.write("Alternative Text: Logo") #### HELPERS @st.cache_data() def make_trues(size: int) -> pd.DataFrame: trues = [True for _ in range(size)] return pd.DataFrame({"true": trues}) #### Construct filter def construct_filters(session_state, mappatura_data): df_true = make_trues(len(mappatura_data)) target_masks = df_true["true"] # level_masks = df_true['true'] objective_masks = df_true["true"] activities_masks = df_true["true"] deprivation_masks = df_true["true"] if len(session_state["filter_target"]) > 0: target_masks = sum( [ mappatura_data["Target"].str.contains(item,regex=False) for item in session_state["filter_target"] ] ) # if len(session_state["filter_level"]) > 0: # level_masks = sum([mappatura_data['Ambito territoriale'].str.contains(item) for item in session_state["filter_level"]]) if len(session_state["filter_objective"]) > 0: objective_masks = sum( [ mappatura_data["Obiettivo sintetico"].str.contains(item,regex=False) for item in session_state["filter_objective"] ] ) if len(session_state["filter_activities"]) > 0: # FIXME: `Attività sitentico` activities_masks = sum( [ mappatura_data["Attività sintetico"].str.contains(item,regex=False) for item in session_state["filter_activities"] ] ) if len(session_state["filter_deprivation"]) > 0: deprivation_masks = sum( [ mappatura_data["Apprendimento/privazione educativa"].str.contains(item,regex=False) for item in session_state["filter_deprivation"] ] ) # full_masks = target_masks & level_masks & objective_masks & activities_masks & deprivation_masks & geo_masks full_masks = ( target_masks & objective_masks & activities_masks & deprivation_masks ) return full_masks def construct_filters_geo(session_state, mappatura_data, st_data): df_true = make_trues(len(mappatura_data)) target_masks = df_true["true"] # level_masks = df_true['true'] objective_masks = df_true["true"] activities_masks = df_true["true"] deprivation_masks = df_true["true"] geo_masks = df_true["true"] if len(session_state["filter_target"]) > 0: target_masks = sum( [ mappatura_data["Target"].str.contains(item, regex=False) for item in session_state["filter_target"] ] ) # if len(session_state["filter_level"]) > 0: # level_masks = sum([mappatura_data['Ambito territoriale'].str.contains(item) for item in session_state["filter_level"]]) if len(session_state["filter_objective"]) > 0: objective_masks = sum( [ mappatura_data["Obiettivo sintetico"].str.contains(item,regex=False) for item in session_state["filter_objective"] ] ) if len(session_state["filter_activities"]) > 0: # FIXME: `Attività sitentico` activities_masks = sum( [ mappatura_data["Attività sintetico"].str.contains(item,regex=False) for item in session_state["filter_activities"] ] ) if len(session_state["filter_deprivation"]) > 0: deprivation_masks = sum( [ mappatura_data["Apprendimento/privazione educativa"].str.contains(item,regex=False) for item in session_state["filter_deprivation"] ] ) last_level = session_state["view"].level if last_level == 1: geo_masks = mappatura_data["Regioni di intervento"].str.contains( session_state["view"].hierarchy_names[-1],regex=False ) if last_level == 2: if ( "last_active_drawing" in st_data and st_data["last_active_drawing"] is not None ): geo_masks = mappatura_data["Comuni di intervento"].str.contains( st_data["last_active_drawing"]["properties"]["name"],regex=False ) else: geo_masks = mappatura_data["Province di intervento"].str.contains( session_state["view"].hierarchy_names[-1],regex=False ) # full_masks = target_masks & level_masks & objective_masks & activities_masks & deprivation_masks & geo_masks full_masks = ( target_masks & objective_masks & activities_masks & deprivation_masks & geo_masks ) return full_masks