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()