277 lines
15 KiB
Python
277 lines
15 KiB
Python
from typing import List, Dict, Any
|
|
import logging
|
|
|
|
from fiscalyear import FiscalYear
|
|
import pandas as pd
|
|
from streamlit.delta_generator import DeltaGenerator
|
|
import streamlit as st
|
|
from constants_module import TRAINING_COUNT_COLUMNS, NEOSERRA_COLUMNS
|
|
|
|
from streamlit_constants import DASHBOARD_CONFIG_OBJECT_KEY
|
|
from utility_classes.base_report_page import BaseReportPage
|
|
from cached_function_wrappers.shared import get_df_centers
|
|
|
|
from cached_function_wrappers.trainings_cached_functions import cached_generate_cleaned_trainings_dataset, cached_generate_center_trainings_count_statistics
|
|
from section_1_graph_library_module import make_network_trainings_count_statistics_charts, StatChartVariants
|
|
from utility_classes.figure_with_max_y import FigureWithMaxY, find_fig_max_y_and_generate_wrapper
|
|
from utility_classes.dashboard_config_parser import DashboardConfig, ExportModulePair
|
|
|
|
|
|
class TrainingsCountStatisticsPage(BaseReportPage):
|
|
"""
|
|
Concrete implementation of a report page analyzing zero-attendee training statistics.
|
|
|
|
This class manages the data pipeline for evaluating un-attended training events across the network.
|
|
It categorizes these events into specific subsets (such as 'First Steps', 'Preplanning', and 'On-Demand')
|
|
to help administrators identify which types of courses historically suffer from zero attendance.
|
|
|
|
:param kwargs: Arbitrary keyword arguments passed to the parent BaseReportPage constructor.
|
|
"""
|
|
|
|
def __init__(self, **kwargs):
|
|
"""
|
|
Initializes temporal filtering boundaries and application configuration state.
|
|
|
|
Captures the current and previous fiscal years to parameterize the report and extracts
|
|
the global dashboard configuration to resolve the necessary Neoserra export endpoints.
|
|
|
|
:param kwargs: Arbitrary keyword arguments.
|
|
"""
|
|
super().__init__("Network Wide Training Counts")
|
|
|
|
self.fiscal_year = FiscalYear.current()
|
|
self.prev_fiscal_year = self.fiscal_year.prev_fiscal_year
|
|
|
|
self.fiscal_year_text = f'FY{str(self.fiscal_year.fiscal_year)[2:]}'
|
|
self.prev_fiscal_year_text = f'FY{str(self.prev_fiscal_year.fiscal_year)[2:]}'
|
|
|
|
# Grab the app config so we can use it to get the export module urls
|
|
self.app_config: DashboardConfig = st.session_state[DASHBOARD_CONFIG_OBJECT_KEY]
|
|
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
def get_fiscal_year_export_url(self, selected_fiscal_year):
|
|
"""
|
|
Resolves the appropriate Neoserra Export Module dataset URL based on the active temporal state.
|
|
|
|
Maps the user-selected fiscal year string to the specific data endpoint, ensuring
|
|
subsequent data fetches query the correct historical or current records.
|
|
|
|
:param selected_fiscal_year: The formatted string representing the user's chosen fiscal year.
|
|
:type selected_fiscal_year: Any
|
|
:return: The URL for the corresponding dataset export.
|
|
:rtype: str
|
|
"""
|
|
export_urls:ExportModulePair = self.app_config.get_trainings_urls()
|
|
if selected_fiscal_year == self.fiscal_year_text:
|
|
return export_urls.current_fy
|
|
else:
|
|
return export_urls.prev_fy
|
|
|
|
@staticmethod
|
|
def get_page_name():
|
|
"""
|
|
Provides the static human-readable identifier for this specific report.
|
|
|
|
Utilized by dashboard orchestrators to populate navigation menus and routing logic.
|
|
|
|
:return: The display name of the report.
|
|
:rtype: str
|
|
"""
|
|
return "Training Counts Statistics"
|
|
|
|
def render_controls(self, container: DeltaGenerator) -> Dict[str, Any]:
|
|
"""
|
|
Defines the user input interface and establishes a safe execution boundary.
|
|
|
|
Renders selection widgets for fiscal year and target centers. Implements a fail-fast
|
|
pattern that halts the Streamlit execution tree if the underlying center dataset fails
|
|
to load, preventing cascading errors during the intensive figure generation phase.
|
|
|
|
:param container: The Streamlit container to attach the input widgets to.
|
|
:type container: DeltaGenerator
|
|
:return: A dictionary containing the user-selected fiscal year and centers.
|
|
:rtype: Dict[str, Any]
|
|
"""
|
|
report_settings_expander = container.expander(
|
|
label="Report Options",
|
|
expanded=True,
|
|
key=self.get_widget_key("report_settings_expander")
|
|
)
|
|
|
|
report_settings_expander.markdown("## Dataset Options")
|
|
report_settings_expander.markdown("These settings will modify the input dataset used to generate the graphs.")
|
|
selected_fiscal_year = report_settings_expander.selectbox(
|
|
label="Fiscal Year",
|
|
options=[self.prev_fiscal_year_text, self.fiscal_year_text],
|
|
index=1,
|
|
key=self.get_widget_key("selected_fiscal_year_selectbox")
|
|
)
|
|
|
|
reportable_only = report_settings_expander.checkbox(label="Reportable only?", value=True, key=self.get_widget_key("reportable_only_checkbox"))
|
|
|
|
include_future_events = report_settings_expander.checkbox(label="Include Future Events?", value=False, key=self.get_widget_key("include_future_events_checkbox"))
|
|
|
|
include_on_demand = report_settings_expander.checkbox(label="Include On-Demand Events?", value=True, key=self.get_widget_key("include_on_demand_checkbox"))
|
|
|
|
export_url = self.get_fiscal_year_export_url(selected_fiscal_year)
|
|
|
|
try:
|
|
all_centers = get_df_centers(export_url)
|
|
except Exception as e:
|
|
self.logger.exception(f"Failed to fetch the dataset for this page: {e}")
|
|
container.error(
|
|
"Failed to get the list of all centers for the dataset for this page. A detailed error message has been added to the logs.")
|
|
st.stop()
|
|
|
|
selected_centers = report_settings_expander.multiselect(label="Centers", options=all_centers,
|
|
default=all_centers,
|
|
key=self.get_widget_key("selected_centers_multiselect"))
|
|
return {
|
|
"selected_fiscal_year":selected_fiscal_year,
|
|
"selected_centers":selected_centers,
|
|
"reportable_only":reportable_only,
|
|
"include_future_events":include_future_events,
|
|
"include_on_demand":include_on_demand,
|
|
}
|
|
|
|
def generate_figures(self, parameters: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Executes the analytical data pipeline and constructs the visualization artifacts.
|
|
|
|
Fetches the trainings dataset and specifically filters for events with zero attendees.
|
|
Generates an extensive suite of paired Plotly charts (absolute count vs. percentage) across
|
|
various course categorizations. Computes strict max-Y bounds for quantity-based charts
|
|
to permit external axis synchronization.
|
|
|
|
:param parameters: The parameter state dictionary captured from the render_controls phase.
|
|
:type parameters: Dict[str, Any]
|
|
:return: A dictionary mapping identifiers to FigureWithMaxY objects and the raw dataframes.
|
|
:rtype: Dict[str, Any]
|
|
"""
|
|
|
|
selected_fiscal_year: str = parameters["selected_fiscal_year"]
|
|
selected_centers: List[str] = parameters["selected_centers"]
|
|
reportable_only:bool = parameters["reportable_only"]
|
|
include_future_events:bool = parameters["include_future_events"]
|
|
include_on_demand:bool = parameters["include_on_demand"]
|
|
|
|
if len(selected_centers) > 0:
|
|
export_url = self.get_fiscal_year_export_url(selected_fiscal_year)
|
|
|
|
try:
|
|
stats_df = cached_generate_center_trainings_count_statistics(export_url, reportable_only=reportable_only, allowed_centers=selected_centers, include_future_events=include_future_events, include_on_demand=include_on_demand)
|
|
trainings_df = cached_generate_cleaned_trainings_dataset(export_url, reportable_only=reportable_only, allowed_centers=selected_centers, include_future_events=include_future_events, include_on_demand=include_on_demand)
|
|
except Exception as e:
|
|
self.logger.exception(f"Failed to generate training count statistics for this page: {e}")
|
|
st.error(f"Failed to generate training count statistics for this page. A detailed error has been added to the logs. {self.app_config.get_errors_contact_string()}")
|
|
st.stop()
|
|
|
|
filter_description = "With 0 Attendees"
|
|
|
|
try:
|
|
collected_figures = make_network_trainings_count_statistics_charts(
|
|
funding_group_df=stats_df,
|
|
filter_description_tag=filter_description,
|
|
fiscal_year_tag=selected_fiscal_year,
|
|
title_prefix="PASBDC*"
|
|
)
|
|
except Exception as e:
|
|
self.logger.exception(f"Failed to build the figures for this report. Got {e}")
|
|
st.error(f"Failed to build the figures for this report. A detailed error has been added to the logs. {self.app_config.get_errors_contact_string()}")
|
|
st.stop()
|
|
|
|
return {
|
|
"total_count": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.TOTAL_COUNT]),
|
|
"total_percent": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.TOTAL_PERCENT]),
|
|
"no_first_steps_count": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.NO_FIRST_STEPS_COUNT]),
|
|
"no_first_steps_percent": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.NO_FIRST_STEPS_PERCENT]),
|
|
"no_first_no_pre_count": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.NO_FIRST_NO_PREPLANNNG_COUNT]),
|
|
"no_first_no_pre_percent": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.NO_FIRST_NO_PREPLANNNG_PERCENT]),
|
|
"first_pre_only_count": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.FIRST_AND_PREPLANNING_ONLY]),
|
|
"first_pre_only_percent": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.FIRST_AND_PREPLANNING_ONLY_PERCENT]),
|
|
"ondemand_count": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.ON_DEMAND_COUNT]),
|
|
"ondemand_percent": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.ON_DEMAND_PERCENT]),
|
|
"ondemand_no_first_count": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.ON_DEMAND_NO_FIRST_STEPS_COUNT]),
|
|
"ondemand_no_first_percent": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.ON_DEMAND_NO_FIRST_STEPS_PERCENT]),
|
|
"ondemand_no_first_no_pre_count": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.ON_DEMAND_NO_FIRST_STEPS_NO_PREPLANNING_COUNT]),
|
|
"ondemand_no_first_no_pre_percent": find_fig_max_y_and_generate_wrapper(collected_figures[StatChartVariants.ON_DEMAND_NO_FIRST_STEPS_NO_PREPLANNING_PERCENT]),
|
|
"stats_df": stats_df,
|
|
"trainings_df": trainings_df
|
|
}
|
|
else:
|
|
st.warning("At least 1 center must be selected!")
|
|
st.stop()
|
|
|
|
|
|
def get_syncable_figure_keys(self) -> List[str]:
|
|
"""
|
|
Declares the specific visualization objects that support dynamic external Y-axis scaling.
|
|
|
|
Explicitly isolates the absolute count charts for synchronization, filtering out all
|
|
percentage-based charts to ensure external axis scaling does not distort proportional views.
|
|
|
|
:return: A list of dictionary keys corresponding to absolute count figures.
|
|
:rtype: List[str]
|
|
"""
|
|
return ["total_count", "no_first_steps_count", "no_first_no_pre_count", "first_pre_only_count", "ondemand_count", "ondemand_no_first_count", "ondemand_no_first_no_pre_count"]
|
|
|
|
def render_figures(self, container: DeltaGenerator, output_data: Dict[str, Any]):
|
|
"""
|
|
Iteratively maps the generated paired visual artifacts to the Streamlit layout.
|
|
|
|
Arranges the charts sequentially using a dynamic 2-column layout loop to directly contrast
|
|
absolute zero-attendance counts with their corresponding proportional percentages. Exposes
|
|
the underlying filtered datasets via an expander module to ensure data transparency.
|
|
|
|
:param container: The Streamlit layout container for the visuals.
|
|
:type container: DeltaGenerator
|
|
:param output_data: The dictionary of computed figures and dataframes from generate_figures.
|
|
:type output_data: Dict[str, Any]
|
|
"""
|
|
chart_pairs = [
|
|
("Total Trainings", "total_count", "total_percent"),
|
|
("Excluding First Steps", "no_first_steps_count", "no_first_steps_percent"),
|
|
("Excluding First Steps & Preplanning", "no_first_no_pre_count", "no_first_no_pre_percent"),
|
|
("First Steps & Preplanning Only", "first_pre_only_count", "first_pre_only_percent"),
|
|
("On-Demand Trainings", "ondemand_count", "ondemand_percent"),
|
|
("On-Demand (Excluding First Steps)", "ondemand_no_first_count", "ondemand_no_first_percent"),
|
|
("On-Demand (Excluding First Steps & Preplanning)", "ondemand_no_first_no_pre_count",
|
|
"ondemand_no_first_no_pre_percent"),
|
|
]
|
|
|
|
stats_df:pd.DataFrame = output_data.get("stats_df")
|
|
trainings_df:pd.DataFrame = output_data.get("trainings_df")
|
|
|
|
for title, count_key, percent_key in chart_pairs:
|
|
# Add a subheader for the section
|
|
container.subheader(title)
|
|
|
|
# Create a 2-column layout
|
|
col1, col2 = container.columns(2)
|
|
|
|
# Extract and render the count figure
|
|
count_fig = output_data.get(count_key)['figure']
|
|
if count_fig:
|
|
col1.plotly_chart(count_fig, use_container_width=True, key=self.get_widget_key(count_key))
|
|
|
|
# Extract and render the percentage figure
|
|
percent_fig = output_data.get(percent_key)['figure']
|
|
if percent_fig:
|
|
col2.plotly_chart(percent_fig, use_container_width=True, key=self.get_widget_key(percent_key))
|
|
|
|
# Add a horizontal line to separate sections cleanly
|
|
container.divider()
|
|
|
|
dataset_expander = container.expander(
|
|
label="Source Datasets",
|
|
expanded=True,
|
|
key=self.get_widget_key("dataset_expander")
|
|
)
|
|
|
|
dataset_expander.markdown("## Source Data")
|
|
dataset_expander.markdown("### Neoserra Trainings Dataset")
|
|
dataset_expander.write(trainings_df)
|
|
dataset_expander.markdown("### Calculated Center Statistics dataset")
|
|
dataset_expander.markdown("Input dataset was filtered to show only 0 attendee events.")
|
|
dataset_expander.write(stats_df) |