import os import time from datetime import datetime import uuid import folium import pandas as pd import streamlit as st from huggingface_hub import HfApi from streamlit_folium import st_folium from src.text_content import ( COLOR_MAPPING, CREDITS_TEXT, HEADERS_MAPPING, ICON_MAPPING, INTRO_TEXT_AR, INTRO_TEXT_EN, INTRO_TEXT_FR, LOGO, REVIEW_TEXT, SLOGAN, ) from src.utils import add_latlng_col, init_map, parse_gg_sheet, is_request_in_list, marker_request from src.map_utils import get_legend_macro TOKEN = os.environ.get("HF_TOKEN", None) VERIFIED_REQUESTS_URL = ( "https://docs.google.com/spreadsheets/d/1PXcAtI5L95hHSXAiRl3Y4v5O4coG39S86OTfBEcvLTE/edit#gid=0" ) REQUESTS_URL = "https://docs.google.com/spreadsheets/d/1gYoBBiBo1L18IVakHkf3t1fOGvHWb23loadyFZUeHJs/edit#gid=966953708" INTERVENTIONS_URL = ( "https://docs.google.com/spreadsheets/d/1eXOTqunOWWP8FRdENPs4cU9ulISm4XZWYJJNR1-SrwY/edit#gid=2089222765" ) api = HfApi(TOKEN) # Initialize Streamlit Config st.set_page_config( layout="wide", initial_sidebar_state="collapsed", page_icon="🤝", page_title="Nt3awnou نتعاونو", ) # Initialize States if "sleep_time" not in st.session_state: st.session_state.sleep_time = 2 if "auto_refresh" not in st.session_state: st.session_state.auto_refresh = False auto_refresh = st.sidebar.checkbox("Auto Refresh?", st.session_state.auto_refresh) if auto_refresh: number = st.sidebar.number_input("Refresh rate in seconds", value=st.session_state.sleep_time) st.session_state.sleep_time = number # Streamlit functions def display_interventions(interventions_df, selected_statuses): """Display NGO interventions on the map""" global fg for index, row in interventions_df.iterrows(): village_status = row[interventions_df.columns[7]] is_future_intervention = ( row[interventions_df.columns[5]] == "Intervention prévue dans le futur / Planned future intervention" ) if pd.isna(village_status) and not is_future_intervention: village_status = "Partiellement satisfait / Partially Served" if village_status not in selected_statuses: continue if is_future_intervention: color_mk = "pink" status = "Planned ⌛" elif village_status != "Critique, Besoin d'aide en urgence / Critical, in urgent need of help": # past intervention and village not in a critical condition color_mk = "green" status = "Done ✅" else: color_mk = "darkgreen" status = "Partial ⚠️" intervention_type = row[interventions_df.columns[6]] org = row[interventions_df.columns[1]] contact = row[interventions_df.columns[2]] city = row[interventions_df.columns[9]] date = row[interventions_df.columns[4]] population = row[interventions_df.columns[11]] details = row[interventions_df.columns[8]] road_state = row[interventions_df.columns[12]] # intervention_info = f"Intervention Status: {status}
Village Status: {village_status}
Org: {org}
Intervention: {intervention_type}
Population: {population}
📅 Date: {date}" intervention_info=f""" Date: {date}
City: {city}
Intervention Status: {status}
Village Status: {village_status}
Org: {org}
Intervention: {intervention_type}
Population: {population}
Road State: {road_state}
Details: {details}
Contact: {contact}
""" if row["latlng"] is None: continue fg.add_child( folium.Marker( location=row["latlng"], tooltip=city, popup=folium.Popup(intervention_info, max_width=300), icon=folium.Icon(color=color_mk), ) ) def show_requests(filtered_df): """Display victim requests on the map""" global fg for index, row in filtered_df.iterrows(): request_type = row["ما هي احتياجاتك؟ (أضفها إذا لم يتم ذكرها)"] displayed_request = marker_request(request_type) # TODO: the marker should depend on selected_options long_lat = row["latlng"] maps_url = f"https://maps.google.com/?q={long_lat}" # we display all requests in popup text and use the first one for the icon/color display_text = f'Request Type: {request_type}
Id: {row["id"]}
Google Maps' icon_name = ICON_MAPPING.get(request_type, "list") if long_lat is None: continue fg.add_child( folium.Marker( location=long_lat, tooltip=row[" لأي جماعة / قيادة / دوار تنتمون ؟"] if not pd.isna(row[" لأي جماعة / قيادة / دوار تنتمون ؟"]) else None, popup=folium.Popup(display_text, max_width=300), icon=folium.Icon( color=COLOR_MAPPING.get(displayed_request, "beige"), icon=icon_name, prefix="glyphicon" ), ) ) def show_verified_requests(filtered_verified_df): """Display verified victim requests on the map""" global fg verified_color_mapping = { "Low": "beige", "Medium": "orange", "High": "red", } for index, row in filtered_verified_df.iterrows(): long_lat = row["latlng"] # we display all requests in popup text and use the first one for the icon/color display_text = "" for col, val in zip(filtered_verified_df.columns, row): if col == "Help Details": request_type = row["Help Details"] marker_request(request_type) # TODO: the marker should depend on selected_options display_text += f"Request Type: {request_type}
" elif col == "Location Details": display_text += f"Location: {val}
" elif col == "Emergency Degree": display_text += f"Emergency Degree: {val}
" elif col == "Verification Date": display_text += f"Verification Date: {val}
" elif col == "id": display_text = f"Id: {val}
" + display_text elif col == "latlng": maps_url = f"https://maps.google.com/?q={val}" display_text += ( f'Google Maps
' ) icon_name = ICON_MAPPING.get(request_type, "list") emergency = row.get("Emergency Degree", "Low") if long_lat is None: continue location = row["Location Details"] fg.add_child( folium.Marker( location=long_lat, tooltip=location if not pd.isna(location) else None, popup=folium.Popup(display_text, max_width=300), icon=folium.Icon( color=verified_color_mapping.get(emergency, "beige"), icon=icon_name, prefix="glyphicon" ), ) ) def display_google_sheet_tables(data_url): """Display the google sheet tables for requests and interventions""" st.markdown( f"""""", unsafe_allow_html=True, ) def display_dataframe(df, drop_cols, data_url, search_id=True, status=False, for_help_requests=False, show_link=True): """Display the dataframe in a table""" col_1, col_2 = st.columns([1, 1]) with col_1: query = st.text_input("🔍 Search for information / بحث عن المعلومات", key=f"query_{uuid.uuid4()}") with col_2: if search_id: id_number = st.number_input( "🔍 Search for an id / بحث عن رقم", min_value=0, max_value=len(filtered_df), value=0, step=1, key=f"id_{uuid.uuid4()}", ) if status: selected_status = st.selectbox( "🗓️ Status / حالة", ["all / الكل", "Done / تم", "Planned / مخطط لها"], key=f"status_{uuid.uuid4()}" ) if query: # Filtering the dataframe based on the query mask = df.apply(lambda row: row.astype(str).str.contains(query).any(), axis=1) display_df = df[mask] else: display_df = df display_df = display_df.drop(drop_cols, axis=1) if search_id and id_number: display_df = display_df[display_df["id"] == id_number] if status: target = "Pouvez-vous nous préciser si vous êtes déjà intervenus ou si vous prévoyez de le faire | Tell us if you already made the intervention, or if you're planning to do it" if selected_status == "Done / تم": display_df = display_df[display_df[target] == "Intervention déjà passée / Past intevention"] elif selected_status == "Planned / مخطط لها": display_df = display_df[display_df[target] != "Intervention déjà passée / Past intevention"] st.dataframe(display_df, height=500) # Original link to the Google Sheet if show_link: st.markdown( f"To view the full Google Sheet for advanced filtering go to: {data_url} **لعرض الورقة كاملة، اذهب إلى**" ) # if we want to check hidden contact information if for_help_requests: st.markdown( "We are hiding contact information to protect the privacy of the victims. If you are an NGO and want to contact the victims, please contact us at nt3awnoumorocco@gmail.com", ) st.markdown( """
nt3awnoumorocco@gmail.com نحن نخفي معلومات الاتصال لحماية خصوصية الضحايا. إذا كنت جمعية وتريد الاتصال بالضحايا، يرجى الاتصال بنا على
""", unsafe_allow_html=True, ) def id_review_submission(): """Id review submission form""" # collapse the text with st.expander("🔍 Review of requests | مراجعة طلب مساعدة"): st.markdown(REVIEW_TEXT) id_to_review = st.number_input("Enter id / أدخل الرقم", min_value=0, max_value=len(df), value=0, step=1) reason_for_review = st.text_area("Explain why / أدخل سبب المراجعة") if st.button("Submit / أرسل"): if reason_for_review == "": st.error("Please enter a reason / الرجاء إدخال سبب") else: filename = f"review_id_{id_to_review}_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.txt" with open(filename, "w") as f: f.write(f"id: {id_to_review}, explanation: {reason_for_review}\n") api.upload_file( path_or_fileobj=filename, path_in_repo=filename, repo_id="nt3awnou/review_requests", repo_type="dataset", ) st.success("Submitted at https://huggingface.co/datasets/nt3awnou/review_requests/ تم الإرسال") # Logo and Title st.markdown(LOGO, unsafe_allow_html=True) # st.title("Nt3awnou نتعاونو") st.markdown(SLOGAN, unsafe_allow_html=True) m = init_map() fg = folium.FeatureGroup(name="Markers") # Selection of requests options = [ "إغاثة", "مساعدة طبية", "مأوى", "طعام وماء", "مخاطر (تسرب الغاز، تلف في الخدمات العامة...)", ] selected_options = [] col1, col2 = st.columns([1, 1]) with col1: show_unverified = st.checkbox( "Display unverified requests / عرض الطلبات غير المؤكدة / Afficher les demandes non vérifiées", value=False, ) with col2: show_interventions = st.checkbox( "Display Interventions | Afficher les interventions | عرض عمليات المساعدة", value=True, ) st.markdown("👉 **Choose request type | Choissisez le type de demande | اختر نوع الطلب**") col1, col2, col3, col4, col5 = st.columns([2, 4, 2, 3, 2]) cols = [col1, col2, col3, col4, col5] for i, option in enumerate(options): checked = cols[i].checkbox(HEADERS_MAPPING[option], value=True) if checked: selected_options.append(option) # Load data and initialize map with plugins df = parse_gg_sheet(REQUESTS_URL) if show_unverified: df = add_latlng_col(df, process_column=15) len_requests = len(df) interventions_df = parse_gg_sheet(INTERVENTIONS_URL) interventions_df = add_latlng_col(interventions_df, process_column="Automatic Extracted Coordinates") len_interventions = len(interventions_df) verified_df = parse_gg_sheet(VERIFIED_REQUESTS_URL) verified_df = add_latlng_col(verified_df, process_column="Automatic Extracted Coordinates") len_verified_requests = len(verified_df) df["id"] = df.index # Needed to display request id verified_df["id"] = verified_df.index # Needed to display request id # keep rows with at least one request in selected_options filtered_df = df[ df["ما هي احتياجاتك؟ (أضفها إذا لم يتم ذكرها)"].apply(lambda x: is_request_in_list(x, selected_options, options)) ] filtered_verified_df = verified_df[ verified_df["Help Details"].apply(lambda x: is_request_in_list(x, selected_options, options)) ] # Selection of interventions st.markdown( "👉 **State of villages visited by NGOs| Etat de villages visités par les ONGs | وضعية القرى التي زارتها الجمعيات**", unsafe_allow_html=True, ) col_1, col_2, col_3 = st.columns([1, 1, 1]) critical_villages = col_1.checkbox( "🚨 Critical, in urgent need of help / وضع حرج، في حاجة عاجلة للمساعدة", value=True, ) partially_satisfied_villages = col_2.checkbox( "⚠️ Partially served / مساعدة جزئية، بحاجة للمزيد من التدخلات", value=True, ) fully_satisfied_villages = col_3.checkbox( "✅ Fully served / تمت المساعدة بشكل كامل", value=True, ) selected_village_types = [] if critical_villages: selected_village_types.append("🚨 Critical, in urgent need of help / وضع حرج، في حاجة عاجلة للمساعدة") if partially_satisfied_villages: selected_village_types.append("⚠️ Partially served / مساعدة جزئية، بحاجة للمزيد من التدخلات") if fully_satisfied_villages: selected_village_types.append("✅ Fully served / تمت المساعدة بشكل كامل") status_mapping = { "🚨 Critical, in urgent need of help / وضع حرج، في حاجة عاجلة للمساعدة": "Critique, Besoin d'aide en urgence / Critical, in urgent need of help", "⚠️ Partially served / مساعدة جزئية، بحاجة للمزيد من التدخلات": "Partiellement satisfait / Partially Served", "✅ Fully served / تمت المساعدة بشكل كامل": "Entièrement satisfait / Fully served", } selected_statuses = [status_mapping[status] for status in selected_village_types] if show_interventions: display_interventions(interventions_df, selected_statuses) # Show requests if show_unverified: show_requests(filtered_df) # Show verified requests show_verified_requests(verified_df) # Add legend legend_macro = get_legend_macro(show_unverified) print(legend_macro) # delete old legend for child in m.get_root()._children: pass # if child.startswith("macro_element"): # m.get_root()._children.remove(child) m.get_root().add_child(legend_macro) print(m.get_root()._children) st_folium(m, use_container_width=True, returned_objects=[], feature_group_to_add=fg, key="map") # Embed code with st.expander("💻 For Developers only, embed code for the map | للمطورين فقط، يمكنك نسخ كود الخريطة"): st.code( """ """, language="html", ) tab_ar, tab_en, tab_fr = st.tabs(["العربية", "English", "Français"]) with tab_en: st.markdown(INTRO_TEXT_EN, unsafe_allow_html=True) col1, col2 = st.columns([1, 1]) with col1: st.markdown( f"""

Number of help requests

{len_requests}

""", unsafe_allow_html=True, ) with col2: st.markdown( f"""

Number of interventions

{len_interventions}

""", unsafe_allow_html=True, ) with tab_ar: st.markdown(INTRO_TEXT_AR, unsafe_allow_html=True) col1, col2 = st.columns([1, 1]) with col1: st.markdown( f"""

عدد طلبات المساعدة

{len_requests}

""", unsafe_allow_html=True, ) with col2: st.markdown( f"""

عدد التدخلات

{len_interventions}

""", unsafe_allow_html=True, ) with tab_fr: st.markdown(INTRO_TEXT_FR, unsafe_allow_html=True) col1, col2 = st.columns([1, 1]) with col1: st.markdown( f"""

Nombre de demandes d'aide

{len_requests}

""", unsafe_allow_html=True, ) with col2: st.markdown( f"""

Nombre d'interventions

{len_interventions}

""", unsafe_allow_html=True, ) # Verified Requests table st.divider() st.subheader("📝 **Table of verified requests / جدول الطلبات المؤكدة**") drop_cols = [ "Phone Number", "id", "Status", "Intervenant ", "Intervention Date", "Any remarks", "VerificationStatus", "Automatic Extracted Coordinates", ] display_dataframe( verified_df, drop_cols, VERIFIED_REQUESTS_URL, search_id=True, for_help_requests=True, show_link=False ) # Requests table st.divider() st.subheader("📝 **Table of requests / جدول الطلبات**") drop_cols = [ "(عند الامكان) رقم هاتف شخص موجود في عين المكان", "الرجاء الضغط على الرابط التالي لمعرفة موقعك إذا كان متاحا", "GeoStamp", "GeoCode", "GeoAddress", "Status", "id", ] display_dataframe(filtered_df, drop_cols, REQUESTS_URL, search_id=True, for_help_requests=True) # Interventions table st.divider() st.subheader("📝 **Table of interventions / جدول التدخلات**") display_dataframe( interventions_df, [], # We show NGOs contact information INTERVENTIONS_URL, search_id=False, status=True, for_help_requests=False, ) # Submit an id for review st.divider() id_review_submission() # Donations can be made to the gouvernmental fund under the name st.divider() st.subheader("📝 **Donations / التبرعات / Dons**") tab_ar, tab_en, tab_fr = st.tabs(["العربية", "English", "Français"]) with tab_en: st.markdown( """

The official bank account dedicated to tackle the consequences of the earthquake is:

Account number:

126

RIB: 001-810-0078000201106203-18
For the money transfers coming from outside Morocco
IBAN: MA64001810007800020110620318
""", unsafe_allow_html=True, ) with tab_ar: st.markdown( """

الحساب البنكي الرسمي المخصص لمواجهة عواقب الزلزال

رقم الحساب

126

RIB: 001-810-0078000201106203-18
للتحويلات القادمة من خارج المغرب
IBAN: MA64001810007800020110620318
""", unsafe_allow_html=True, ) with tab_fr: st.markdown( """

Le compte bancaire officiel dédié à la lutte contre les conséquences du séisme est le suivant:

Numéro de compte:

126

RIB: 001-810-0078000201106203-18
Pour les transferts d'argent en provenance de l'étranger
IBAN: MA64001810007800020110620318
""", unsafe_allow_html=True, ) # Credits st.markdown( CREDITS_TEXT, unsafe_allow_html=True, ) if auto_refresh: time.sleep(number) st.experimental_rerun()