utils.py

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