265 lines
13 KiB
Python
265 lines
13 KiB
Python
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(
|
|
"<b>NOTE: Documentation levels were determined as follows.</b><br>"
|
|
"<b>Documented: Will be submitted to Nexus as long as 'Director Verified is checked'</b></br>There is a non-blank, non-'Requested on eCenter' attribution source AND Affirmation Statement was non-blank</br>"
|
|
"<b>Affirmation Statement Missing: Will NOT be submitted to Nexus</b></br>Attribution source is non-blank, non-'Requested on eCenter' BUT affirmation statement was blank.</br>"
|
|
"<b>Not Documented: Will NOT be submitted to Nexus</b></br> 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"] |