Spaces:
Running
Running
import os | |
import time | |
import requests | |
import folium | |
import pandas as pd | |
from datetime import datetime | |
import streamlit as st | |
from streamlit_folium import st_folium | |
from utils import legend_macro | |
from huggingface_hub import HfApi | |
TOKEN = os.environ.get("HF_TOKEN", None) | |
api = HfApi(token=TOKEN) | |
st.set_page_config(layout="wide", initial_sidebar_state="collapsed") | |
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 | |
session = requests.Session() | |
def parse_latlng_from_link(url): | |
try: | |
# extract latitude and longitude from gmaps link | |
if "@" not in url: | |
# We first need to get the redirect URL | |
resp = session.head(url, allow_redirects=True) | |
url = resp.url | |
latlng = url.split('@')[1].split(',')[0:2] | |
return [float(latlng[0]), float(latlng[1])] | |
except Exception as e: | |
print(f"Error parsing latlng from link: {e}") | |
return None | |
def parse_gg_sheet_interventions(url): | |
df = pd.read_csv(url) | |
return df.assign(latlng=df.iloc[:, 3].apply(parse_latlng_from_link)) | |
def parse_gg_sheet(url): | |
url = url.replace("edit#gid=", "export?format=csv&gid=") | |
df = pd.read_csv(url) | |
# parse latlng (column 4) to [lat, lng] | |
def parse_latlng(latlng): | |
try: | |
lat, lng = latlng.split(",") | |
return [float(lat), float(lng)] | |
except Exception as e: | |
print(f"Error parsing latlng: {e}") | |
return None | |
return df.assign(latlng=df.iloc[:, 4].apply(parse_latlng)) | |
df = parse_gg_sheet( | |
"https://docs.google.com/spreadsheets/d/1gYoBBiBo1L18IVakHkf3t1fOGvHWb23loadyFZUeHJs/edit#gid=966953708" | |
) | |
interventions_df = parse_gg_sheet_interventions( | |
"https://docs.google.com/spreadsheets/d/1eXOTqunOWWP8FRdENPs4cU9ulISm4XZWYJJNR1-SrwY/gviz/tq?tqx=out:csv" | |
) | |
# select requests | |
headers_mapping = { | |
"إغاثة": "Rescue/إغاثة", | |
"مساعدة طبية": "Medical Assistance/مساعدة طبية", | |
"مأوى": "Shelter/مأوى", | |
"طعام وماء": "Food & Water/طعام وماء", | |
"مخاطر (تسرب الغاز، تلف في الخدمات العامة...)": "Danger/مخاطر (تسرب الغاز، تلف في الخدمات العامة...)", | |
} | |
colors_mapping = { | |
"إغاثة": "red", | |
"مساعدة طبية": "orange", | |
"مأوى": "beige", | |
"طعام وماء": "blue", | |
"مخاطر (تسرب الغاز، تلف في الخدمات العامة...)": "gray", | |
} | |
icon_mapping = { | |
"إغاثة": "bell", # life ring icon for rescue | |
"مساعدة طبية": "heart", # medical kit for medical assistance | |
"مأوى": "home", # home icon for shelter | |
"طعام وماء": "cutlery", # cutlery (fork and knife) for food & water | |
"مخاطر (تسرب الغاز، تلف في الخدمات العامة...)": "Warning" # warning triangle for dangers | |
} | |
options = ["إغاثة", "مساعدة طبية", "مأوى", "طعام وماء", "مخاطر (تسرب الغاز، تلف في الخدمات العامة...)"] | |
selected_options = [] | |
st.markdown("👉 **Choose request type / اختر نوع الطلب**") | |
col1, col2, col3, col4, col5 = st.columns([2, 3, 2, 3, 4]) | |
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(headers_mapping[option]) | |
arabic_options = [e.split("/")[1] for e in selected_options] | |
df['id'] = df.index | |
filtered_df = df[df['ما هي احتياجاتك؟ (أضفها إذا لم يتم ذكرها)'].isin(arabic_options)] | |
selected_headers = [headers_mapping[request] for request in arabic_options] | |
# select interventions | |
st.markdown("👇 **View past or planned interventions / عرض عمليات المساعدة السابقة أو المخطط لها**") | |
#show_interventions = st.checkbox("Display Interventions عرض التدخلات", value=True) | |
show_interventions = True | |
m = folium.Map( | |
location=[31.228674, -7.992047], | |
zoom_start=8.5, | |
min_zoom=8.5, | |
max_lat=35.628674, | |
min_lat=29.628674, | |
max_lon=-4.992047, | |
min_lon=-10.992047, | |
max_bounds=True, | |
) | |
if show_interventions: | |
for index, row in interventions_df.iterrows(): | |
status = "Done ✅" if row[interventions_df.columns[5]]!="Intervention prévue dans le futur / Planned future intervention" else "Planned ⌛" | |
color_mk = "green" if row[interventions_df.columns[5]]!="Intervention prévue dans le futur / Planned future intervention" else "pink" | |
intervention_type = row[interventions_df.columns[6]].split("/")[0].strip() | |
org = row[interventions_df.columns[1]] | |
city = row[interventions_df.columns[9]] | |
date = row[interventions_df.columns[4]] | |
intervention_info = f"<b>Status:</b> {status}<br><b>Org:</b> {org}<br><b>Intervention:</b> {intervention_type}<br><b>📅 Date:</b> {date}" | |
if row["latlng"] is None: | |
continue | |
folium.Marker( | |
location=row["latlng"], | |
tooltip=city, | |
popup=folium.Popup(intervention_info, max_width=300), | |
icon=folium.Icon(color=color_mk) | |
).add_to(m) | |
for index, row in filtered_df.iterrows(): | |
request_type = row['ما هي احتياجاتك؟ (أضفها إذا لم يتم ذكرها)'] | |
display_text = f"<b>Request Type:</b> {request_type}<br><b>Id:</b> {row['id']}" | |
icon_name = icon_mapping.get(request_type, 'info-sign') | |
if row["latlng"] is None: | |
continue | |
folium.Marker( | |
location=row["latlng"], | |
tooltip=row[' لأي جماعة / قيادة / دوار تنتمون ؟'] if not pd.isna(row[' لأي جماعة / قيادة / دوار تنتمون ؟']) else None, | |
popup=folium.Popup(display_text, max_width=300), | |
icon=folium.Icon(color=colors_mapping.get(request_type, "blue"), icon=icon_name) | |
).add_to(m) | |
# Macro to add legend | |
m.get_root().add_child(legend_macro) | |
st_data = st_folium(m, use_container_width=True) | |
if auto_refresh: | |
time.sleep(number) | |
st.experimental_rerun() |