from typing import Dict, Any, List import logging import os import pandas as pd import streamlit as st from utility_classes.base_report_page import BaseReportPage from utility_classes.figure_with_max_y import FigureWithMaxY from cached_function_wrappers.shared import get_df_centers from cached_function_wrappers.funding_milestones_cached_functions import cached_sanitize_funding_data from streamlit_constants import DASHBOARD_CONFIG_OBJECT_KEY from utility_classes.dashboard_config_parser import DashboardConfig, ExportModulePair from section_1_graph_library_module import ( make_funding_attribution_network_wide, make_funding_attribution_rate_chart, make_theoretical_funding_attribution_rate_chart ) from constants_module import NEOSERRA_COLUMNS, OUT_COLUMNS from streamlit.delta_generator import DeltaGenerator from fiscalyear import * from plotly.graph_objects import Figure class NetworkFundingMilestonesReportPage(BaseReportPage): """ Concrete implementation of a report page for analyzing network-wide funding milestones. This class manages the data pipeline and UI rendering for funding metrics. It handles fiscal year temporal states, retrieves specific data export URLs from the global dashboard configuration, and generates targeted attribution charts to evaluate funding documentation rates. :param kwargs: Arbitrary keyword arguments passed to the parent constructor. """ def __init__(self, **kwargs): """ Initializes the report's temporal context and environmental state. Extracts current and previous fiscal year boundaries and retrieves the global dashboard configuration from the Streamlit session state to locate the necessary data export URLs. :param kwargs: Arbitrary keyword arguments. """ super().__init__("Network Milestones Report") 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:]}' def get_fiscal_year_export_url(self, selected_fiscal_year): """ Resolves the appropriate external dataset URL based on user selection. Maps the selected fiscal year string from the UI to the correct data endpoint configured in the application environment. :param selected_fiscal_year: The formatted string representing the user's chosen fiscal year. :type selected_fiscal_year: str :return: The URL for the corresponding dataset export. :rtype: str """ export_urls:ExportModulePair = self.app_config.get_funding_milestones_urls() if selected_fiscal_year == self.fiscal_year_text: return export_urls.current_fy else: return export_urls.prev_fy return self.current_fy_export_url if selected_fiscal_year == self.fiscal_year_text else self.previous_fy_export_url @staticmethod def get_page_name(): """ Provides the static human-readable identifier for this report. Used by the page comparer to route users to this specific page. :return: The display name of the report. :rtype: str """ return "Network Funding Milestones" def render_controls(self, container: DeltaGenerator) -> Dict[str, Any]: """ Defines and captures the user input state required to filter the funding report. Renders selection widgets for fiscal year and center filtering. It fetches the required list of centers dynamically and halts the Streamlit execution sequence if the dataset fails to load to prevent downstream rendering errors. :param container: The Streamlit container to draw the widgets onto. :type container: DeltaGenerator :return: A dictionary containing 'selected_fiscal_year' and 'selected_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")) 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( f"Failed to get the list of all centers for the dataset for this page. A detailed error message has been added to the logs. {self.app_config.get_errors_contact_string()}") 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 } def generate_figures(self, parameters: Dict[str, Any]): """ Executes the core data processing and visualization pipeline for funding milestones. Fetches the targeted dataset, processes the funding metrics, and constructs the Plotly figure objects. It also computes absolute maximums for necessary charts to allow parent orchestrators to synchronize axes across multiple instances. :param parameters: The dictionary of user inputs captured from render_controls. :type parameters: Dict[str, Any] :return: A dictionary containing FigureWithMaxY objects for the charts and the raw dataframe. :rtype: Dict[str, Any] """ selected_fiscal_year = parameters['selected_fiscal_year'] selected_centers = parameters['selected_centers'] reportable_only:bool = parameters['reportable_only'] export_url = self.get_fiscal_year_export_url(selected_fiscal_year) try: funding_df = cached_sanitize_funding_data(export_url, reportable_only=reportable_only, allowed_centers=selected_centers) except Exception as e: self.logger.exception(f"Failed to fetch the dataset for this page. Got {e}") st.error(f"Failed to fetch the dataset for this page. A detailed error has been added to the logs. {self.app_config.get_errors_contact_string()}") st.stop() try: network_wide_funding_fig = make_funding_attribution_network_wide( funding_df, graph_note="", fiscal_year=selected_fiscal_year, title="Capital Funding Attributions Per Center", network_label="PASBDC*", col_neo_center=NEOSERRA_COLUMNS.center, col_documentation_level=OUT_COLUMNS.milestone_documentation_level ) except Exception as e: self.logger.exception(f"Failed to generate the network wide funding figure for this page. Got {e}") st.error(f"Failed to generate the network wide funding figure for this page. A detailed error has been added to the logs. {self.app_config.get_errors_contact_string()}") st.stop() try: documented_only_fig = make_funding_attribution_rate_chart( funding_df, fiscal_year=selected_fiscal_year, documented_tag=OUT_COLUMNS.val_documented, col_neo_center=NEOSERRA_COLUMNS.center, col_documentation_level=OUT_COLUMNS.milestone_documentation_level ) except Exception as e: self.logger.exception(f"Failed to generate the documented only figure for this page. Got {e}") st.error( f"Failed to generate the documented only figure for this page. A detailed error has been added to the logs. {self.app_config.get_errors_contact_string()}") st.stop() try: theoretical_fig = make_theoretical_funding_attribution_rate_chart( funding_df, title="Documented Percentage if All Funding Milestones With an Attribution Source had an Affirmation", fiscal_year=selected_fiscal_year, documented_tag=OUT_COLUMNS.val_documented, affirmation_missing_tag=OUT_COLUMNS.val_affirmation_missing, col_neo_center=NEOSERRA_COLUMNS.center, col_documentation_level=OUT_COLUMNS.milestone_documentation_level ) except Exception as e: self.logger.exception(f"Failed to generate the theoretical attribution figure for this page. Got {e}") st.error( f"Failed to generate the theoretical attribution figure for this page. A detailed error has been added to the logs. {self.app_config.get_errors_contact_string()}") st.stop() # Obtain the max y value for the chart we care about synchronizing the axis with if this report is in # a page comparer max_bar_y = funding_df[NEOSERRA_COLUMNS.center].value_counts().max() return { "network_wide_funding_fig":FigureWithMaxY(figure=network_wide_funding_fig, max_y=max_bar_y), "documented_only_fig":FigureWithMaxY(figure=documented_only_fig, max_y=0.0), "theoretical_fig":FigureWithMaxY(figure=theoretical_fig, max_y=0.0), "funding_df":funding_df } def render_figures(self, container: DeltaGenerator, output_data: Dict[str, Any]): """ Maps the generated analytical figures and raw data to Streamlit UI components. Draws the instantiated Plotly objects to the container and provides the necessary static context, such as definitions for the documentation levels, to help users interpret the charts. :param container: The Streamlit container to draw the visuals onto. :type container: DeltaGenerator :param output_data: The computed figure objects and dataframe returned by generate_figures. :type output_data: Dict[str, Any] """ network_wide_funding_fig:Figure= output_data['network_wide_funding_fig']['figure'] documented_only_fig:Figure = output_data['documented_only_fig']['figure'] theoretical_fig:Figure = output_data['theoretical_fig']['figure'] funding_df:pd.DataFrame = output_data['funding_df'] container.plotly_chart(network_wide_funding_fig, key=self.get_widget_key("network_wide_funding_fig")) container.markdown( "NOTE: Documentation levels were determined as follows.
" "Documented: Will be submitted to Nexus as long as 'Director Verified is checked'
There is a non-blank, non-'Requested on eCenter' attribution source AND Affirmation Statement was non-blank
" "Affirmation Statement Missing: Will NOT be submitted to Nexus
Attribution source is non-blank, non-'Requested on eCenter' BUT affirmation statement was blank.
" "Not Documented: Will NOT be submitted to Nexus
The attribution source is blank or 'Requested on eCenter'.", unsafe_allow_html=True ) container.plotly_chart(documented_only_fig, key=self.get_widget_key("documented_only_fig")) container.plotly_chart(theoretical_fig, key=self.get_widget_key("theoretical_fig")) 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(funding_df) def get_syncable_figure_keys(self) -> List[str]: """ Declares the figures that support dynamic Y-axis scaling for side-by-side comparisons. Restricts synchronization to the network-wide funding chart, as applying external axis limits to percentage-based or non-quantity charts would distort the data representation. :return: A list containing the dictionary key for the network-wide funding figure. :rtype: List[str] """ return ["network_wide_funding_fig"]