nvtbsod / utils.py
smomtaz's picture
Update utils.py
33509eb verified
Raw
History Blame Contribute Delete
6.25 kB
from pathlib import Path
import pandas as pd
import geopandas as gpd
import faicons as fa
from shiny import ui
from shared import app_dir
import matplotlib.pyplot as plt
import numpy as np
#app_dir = Path(__file__).parent
#tips = pd.read_csv(app_dir / "tips.csv")
def load_data():
"""
Load zones shapefile and trips data.
Returns:
zones (GeoDataFrame): GeoDataFrame of zones.
trips (DataFrame): DataFrame of trip data.
"""
zones = gpd.read_file(app_dir / "data" / "zone" / "Napa_TBS_2024_Zone_System.shp") # Replace with your shapefile path
zones = zones[zones.is_valid & ~zones.is_empty]
zones = zones[["FPID", "geometry"]] # Keep only essential columns
zones["geometry"] = zones["geometry"].simplify(0.001, preserve_topology=True) # Simplify geometries
#trips = pd.read_csv(app_dir / "data" / "od_table.csv") # Replace with your trip data path
trips = pd.read_parquet(app_dir / "data" / "od_table.parquet") # Replace with your trip data path
return zones, trips
def load_excel_data(SHEET_NAME):
"""
Load data tables from Excel
Returns:
data (DataFrame): DataFrame of data table.
"""
EXCEL_FILE = app_dir / "data" / "trips_data.xlsx" # Update with your file path
data = pd.read_excel(EXCEL_FILE, sheet_name=SHEET_NAME)
return data
def filter_trips(trips, zones, origin_filter, destination_filter):
"""
Filter trips based on selected origin or destination.
Args:
trips (DataFrame): Trip data with origin and destination zones.
zones (GeoDataFrame): GeoDataFrame of zones.
origin_filter (str): Selected origin zone ID.
destination_filter (str): Selected destination zone ID.
Returns:
filtered_origins (GeoDataFrame): Filtered origin zones with trip counts.
filtered_destinations (GeoDataFrame): Filtered destination zones with trip counts.
"""
if origin_filter and destination_filter:
filtered_trips = trips[
(trips["start_zone_id"] == origin_filter) &
(trips["end_zone_id"] == destination_filter)
]
elif origin_filter:
filtered_trips = trips[trips["start_zone_id"] == origin_filter]
elif destination_filter:
filtered_trips = trips[trips["end_zone_id"] == destination_filter]
else:
filtered_trips = trips
# Aggregate trip counts for origins and destinations
filtered_origins = zones.merge(
#filtered_trips.groupby("start_zone_id").size().reset_index(name="trip_count"),
filtered_trips.groupby("start_zone_id")["trips"].sum().reset_index(name="trip_count"),
left_on="FPID", right_on="start_zone_id", how="left"
).fillna(0)
filtered_destinations = zones.merge(
filtered_trips.groupby("end_zone_id")["trips"].sum().reset_index(name="trip_count"),
left_on="FPID", right_on="end_zone_id", how="left"
).fillna(0)
return filtered_origins, filtered_destinations
def create_nav_button(button_icon, button_label_text, button_link, button_cls = "btn btn-primary"):
return ui.tags.a(
ui.HTML(f"{fa.icon_svg(button_icon)} {button_label_text}"),
href=button_link,
class_=button_cls
)
def conlogo():
img = {
"src": app_dir / "images" / "logo.png",
"style": "width: 80%; height: auto; max-height: 150px; margin-bottom: 0px;"
}
return img
def vendorlogo():
img = {
"src": app_dir / "images" / "logo2.png",
"style": "width: 80%; height: auto; max-height: 150px; margin-bottom: 0px;"
}
return img
def create_bar_chart(categories, values, plottitle = "Bar Chart Example"):
fig, ax = plt.subplots()
#ax.bar(categories, values, color=["#1f77b4", "#ff7f0e", "#2ca02c"])
ax.bar(categories, values)
ax.set_title(plottitle)
ax.set_ylabel("Values")
ax.set_xlabel("Categories")
return fig
def create_pie_chart(labels, sizes, plottitle):
#sizes = [22, 41, 37]
#colors = ["#00b3b3", "#70d281", "#ff7f0e"]
#explode = (0.1, 0, 0) # Explode the first slice for emphasis
fig, ax = plt.subplots()
wedges, texts, autotexts = ax.pie(
sizes,
#explode=explode,
labels=labels,
autopct="%1.0f%%",
startangle=90,
#colors=colors,
textprops=dict(color="black"),
wedgeprops=dict(width=0.4) # Adjust width for the donut effect
)
# Customizing the labels
for text, label in zip(autotexts, labels):
text.set_color("black")
text.set_fontsize(12)
#ax.set_title("What Types of Trips are Occuring within Napa County on a Weekday?", fontsize=16, fontweight="bold")
ax.set_title(plottitle, fontsize=16, fontweight="bold")
return fig
def create_stacked_bar_chart(categories, intra_napa, into_napa, out_napa):
# Data for the stacked bar chart
#categories = ["Early AM", "AM Peak", "Mid-Day", "PM Peak", "Evening"]
#intra_napa = [2000, 16000, 14000, 12000, 5000]
#into_napa = [1000, 8000, 2000, 3000, 2000]
#out_napa = [2000, 4000, 2000, 5000, 1000]
bar_width = 0.5 # Width of the bars
x = np.arange(len(categories)) # x-axis positions
fig, ax = plt.subplots(figsize=(8, 6))
# Stacking the bars
ax.bar(x, intra_napa, bar_width, label="Intra-Napa County", color="#70d281")
ax.bar(x, into_napa, bar_width, bottom=intra_napa, label="Into Napa County", color="#ff7f0e")
ax.bar(x, out_napa, bar_width, bottom=np.array(intra_napa) + np.array(into_napa), label="Out of Napa County", color="#00b3b3")
# Customizing the plot
ax.set_xticks(x)
ax.set_xticklabels(categories)
ax.set_ylabel("Trips")
ax.set_title("Weekday Work Trip Types")
ax.legend(loc="upper right")
# Add value annotations
for i in range(len(categories)):
ax.text(x[i], intra_napa[i] / 2, f"{intra_napa[i]}", ha="center", va="center", color="white")
ax.text(x[i], intra_napa[i] + into_napa[i] / 2, f"{into_napa[i]}", ha="center", va="center", color="white")
ax.text(x[i], intra_napa[i] + into_napa[i] + out_napa[i] / 2, f"{out_napa[i]}", ha="center", va="center", color="white")
plt.tight_layout()
return fig