from typing import List, Dict, Any import logging import pandas as pd from fiscalyear import FiscalYear from streamlit.delta_generator import DeltaGenerator import streamlit as st from constants_module import TRAINING_COUNT_COLUMNS, NEOSERRA_COLUMNS, OUT_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 from section_1_graph_library_module import ( make_attendee_bins_statistics_charts ) from shared_tools_module import StatChartVariants from utility_classes.figure_with_max_y import find_fig_max_y_and_generate_wrapper, FigureWithMaxY from utility_classes.dashboard_config_parser import DashboardConfig, ExportModulePair class TrainingAttendeeRanges(BaseReportPage): """ Concrete implementation of a report page analyzing training attendee size distributions. This class manages the pipeline for categorizing training events into attendee size brackets (bins). It isolates specific training types—such as 'First Steps' and 'Preplanning'—to evaluate how introductory courses impact the overall network attendee size distributions. :param kwargs: Arbitrary keyword arguments passed to the parent BaseReportPage constructor. """ def __init__(self, **kwargs): """ Initializes the temporal boundaries and configuration state for the attendee ranges report. Captures current and previous fiscal years to manage report filtering and extracts the global application configuration to resolve the appropriate external data endpoints. :param kwargs: Arbitrary keyword arguments. """ super().__init__("Network Wide Training Attendee Ranges") 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 external dataset endpoint based on the active temporal state. Maps the user's selected fiscal year to the appropriate data URL, ensuring the pipeline fetches the correct historical or current training records. :param selected_fiscal_year: The string representation of the 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 display identifier for this report module. Utilized by dashboard orchestrators to construct routing and navigation menus. :return: The human-readable name of the report. :rtype: str """ return "Training Attendee Ranges" 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 strict fail-fast pattern that halts the Streamlit execution sequence if the baseline dataset fails to load, preventing downstream rendering errors. :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 binned visualization objects. Fetches the trainings dataset and generates a suite of chart variants comparing absolute counts against percentages across different training subsets. Computes strict max-Y values for the quantity-based charts to support external dynamic 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 dataframe. :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"] export_url = self.get_fiscal_year_export_url(selected_fiscal_year) 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 ) self.logger.error(f"{trainings_df.info()}") try: bins_figs = make_attendee_bins_statistics_charts( trainings_df, center="Network Wide", network_label="PASBDC*", fiscal_year_tag=selected_fiscal_year, first_steps_vals=['First Steps', 'Next Steps'], preplanning_val=OUT_COLUMNS.val_preplanning, col_neo_attendees_total=NEOSERRA_COLUMNS.attendees_total, col_attendees_range=OUT_COLUMNS.attendees_range, col_neo_primary_topic=NEOSERRA_COLUMNS.primary_training_topic ) except Exception as e: self.logger.exception(f"Failed to generate figures for this page: {e}") st.error(f"Failed to generate the figures for this page. 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(bins_figs[StatChartVariants.TOTAL_COUNT]), 'total_percent':FigureWithMaxY(figure=bins_figs[StatChartVariants.TOTAL_PERCENT], max_y=0.0), 'no_first_no_pre_count':find_fig_max_y_and_generate_wrapper(bins_figs[StatChartVariants.NO_FIRST_NO_PREPLANNNG_COUNT]), 'no_first_no_pre_percent':FigureWithMaxY(figure=bins_figs[StatChartVariants.NO_FIRST_NO_PREPLANNNG_PERCENT], max_y=0.0), 'first_pre_only_count':find_fig_max_y_and_generate_wrapper(bins_figs[StatChartVariants.FIRST_AND_PREPLANNING_ONLY]), 'first_pre_only_percent':FigureWithMaxY(figure=bins_figs[StatChartVariants.FIRST_AND_PREPLANNING_ONLY_PERCENT], max_y=0.0), 'trainings_df':trainings_df } 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 repetitive 2-column layout to directly contrast absolute counts with proportional percentages for each training subset. Exposes the raw underlying dataset via an expander module to ensure data auditing and 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 = [ ("All Trainings", "total_count", "total_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"), ] 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) def get_syncable_figure_keys(self) -> List[str]: """ Declares the specific figures that permit dynamic external Y-axis scaling. Explicitly isolates the absolute count charts for synchronization, filtering out the percentage-based charts to ensure external axis scaling does not distort proportional data. :return: A list of dictionary keys corresponding to absolute count figures. :rtype: List[str] """ return ["total_count", "no_first_no_pre_count", "first_pre_only_count"]