254 lines
11 KiB
Python
254 lines
11 KiB
Python
from typing import Dict, Any, List
|
|
import logging
|
|
|
|
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, find_fig_max_y_and_generate_wrapper, extract_figure_data
|
|
from cached_function_wrappers.shared import get_df_centers
|
|
from cached_function_wrappers.funding_milestones_cached_functions import cached_sanitize_funding_data
|
|
from cached_function_wrappers.nbs_cached_functions import cached_get_nbs_data
|
|
from streamlit_constants import DASHBOARD_CONFIG_OBJECT_KEY
|
|
from utility_classes.dashboard_config_parser import DashboardConfig, ExportModulePair
|
|
from milestone_attribution_graph_library_module import (
|
|
make_attribution_pie,
|
|
make_attribution_grouped_chart
|
|
)
|
|
|
|
from constants_module import NEOSERRA_COLUMNS, OUT_COLUMNS
|
|
|
|
from streamlit.delta_generator import DeltaGenerator
|
|
from fiscalyear import *
|
|
|
|
class CenterMilestonesReportPage(BaseReportPage):
|
|
"""
|
|
Implements a specific report page for visualizing center-level milestones.
|
|
|
|
This class manages the lifecycle for displaying either 'Capital Funding' or
|
|
'New Business Starts' milestones. It calculates current and previous
|
|
fiscal years to provide temporal context and interacts with the Neoserra
|
|
export module to fetch and sanitize center-specific datasets.
|
|
|
|
:param kwargs: Arbitrary keyword arguments.
|
|
"""
|
|
def __init__(self, **kwargs):
|
|
super().__init__("Center Specific Milestones")
|
|
|
|
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:]}'
|
|
|
|
self.capital_funding_report_key = "Capital Funding"
|
|
self.new_business_start_report_key = "New Business Starts"
|
|
|
|
|
|
def get_fiscal_year_export_url(self, selected_fiscal_year:str, is_nbs:bool) -> str:
|
|
"""
|
|
Determines the appropriate data export URL based on report type and year.
|
|
|
|
Logic branches between New Business Start (NBS) and Funding milestones,
|
|
then selects the URL corresponding to the current or previous fiscal
|
|
year stored in the application configuration.
|
|
|
|
:param selected_fiscal_year: The fiscal year string (e.g., 'FY26').
|
|
:type selected_fiscal_year: str
|
|
:param is_nbs: Flag indicating if the report type is New Business Starts.
|
|
:type is_nbs: bool
|
|
:return: The URL for the requested data export.
|
|
:rtype: str
|
|
"""
|
|
if is_nbs:
|
|
export_urls:ExportModulePair = self.app_config.get_nbs_milestones_urls()
|
|
else:
|
|
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
|
|
|
|
@staticmethod
|
|
def get_page_name():
|
|
"""
|
|
Returns the static display name for the report page.
|
|
|
|
:return: "Center Specific Milestones"
|
|
:rtype: str
|
|
"""
|
|
return "Center Specific Milestones"
|
|
|
|
def render_controls(self, container: DeltaGenerator) -> Dict[str, Any]:
|
|
"""
|
|
Renders UI widgets to capture reporting parameters from the user.
|
|
|
|
Displays an expander containing selectors for the target fiscal year,
|
|
milestone type, and specific center. It validates the connection to
|
|
the data source before allowing center selection.
|
|
|
|
:param container: The Streamlit container for control widgets.
|
|
:type container: DeltaGenerator
|
|
:return: Dictionary containing 'selected_fiscal_year', 'selected_center', and 'is_nbs'.
|
|
: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")
|
|
)
|
|
|
|
report_type = report_settings_expander.selectbox(
|
|
label="Milestone Type",
|
|
options=[self.capital_funding_report_key, self.new_business_start_report_key],
|
|
index=0,
|
|
key=self.get_widget_key("report_type_selectbox")
|
|
)
|
|
|
|
reportable_only = report_settings_expander.checkbox(label="Reportable only?", value=True, key=self.get_widget_key("reportable_only_checkbox"))
|
|
|
|
is_nbs = True if report_type == self.new_business_start_report_key else False
|
|
|
|
export_url = self.get_fiscal_year_export_url(selected_fiscal_year, is_nbs)
|
|
|
|
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_center = report_settings_expander.selectbox(
|
|
label="Selected Center",
|
|
options=all_centers,
|
|
index=0,
|
|
key=self.get_widget_key("selected_center_selectbox")
|
|
)
|
|
|
|
return {
|
|
"selected_fiscal_year":selected_fiscal_year,
|
|
"selected_center":selected_center,
|
|
"is_nbs":is_nbs,
|
|
"reportable_only":reportable_only
|
|
}
|
|
|
|
def generate_figures(self, parameters: Dict[str, Any]):
|
|
"""
|
|
Fetches raw milestone data and constructs visualization objects.
|
|
|
|
Coordinates the data retrieval pipeline using cached wrappers and
|
|
processes the resulting DataFrames into attribution pie charts and
|
|
grouped bar charts. Figures are wrapped in utility classes to
|
|
preserve metadata like Y-axis maximums for potential synchronization.
|
|
|
|
:param parameters: User inputs containing center, year, and report type.
|
|
:type parameters: Dict[str, Any]
|
|
:return: Dictionary containing 'pie_fig', 'bar_fig', and 'milestone_df'.
|
|
:rtype: Dict[str, Any]
|
|
"""
|
|
selected_fiscal_year = parameters['selected_fiscal_year']
|
|
selected_center = parameters['selected_center']
|
|
is_nbs = parameters['is_nbs']
|
|
reportable_only = parameters['reportable_only']
|
|
|
|
try:
|
|
export_url:str = self.get_fiscal_year_export_url(selected_fiscal_year, is_nbs)
|
|
if is_nbs:
|
|
milestone_df = cached_get_nbs_data(export_url, reportable_only=reportable_only, allowed_centers=[selected_center])
|
|
else:
|
|
milestone_df = cached_sanitize_funding_data(export_url, reportable_only=reportable_only, allowed_centers=[selected_center])
|
|
except Exception as e:
|
|
self.logger.exception(f"Failed to fetch the dataset for this page: {e}")
|
|
st.error(
|
|
f"An issue was encountered while fetching the data for this page. A detailed error message has been recorded in the logs. {self.app_config.get_errors_contact_string()}")
|
|
st.stop()
|
|
|
|
title_tag = "New Business Start" if is_nbs else "Funding"
|
|
try:
|
|
pie_fig = make_attribution_pie(
|
|
milestone_df,
|
|
title=f"{selected_center} Documented vs. Not Documented {title_tag} Milestones {selected_fiscal_year}",
|
|
date_note=datetime.datetime.now().strftime('%m/%d/%Y'),
|
|
col_documentation_level=OUT_COLUMNS.milestone_documentation_level
|
|
)
|
|
except Exception as e:
|
|
self.logger.exception(f"Failed to generate the pie chart graphic with error: {e}")
|
|
st.error(f"An issue was encountered while generating the pie chart graphic. A detailed error message has been recorded in the logs. {self.app_config.get_errors_contact_string()}")
|
|
st.stop()
|
|
|
|
try:
|
|
bar_fig = make_attribution_grouped_chart(
|
|
milestone_df,
|
|
title=f"{selected_center} Attribution Source vs. Documentation Level For {title_tag} Milestones {selected_fiscal_year}",
|
|
)
|
|
bar_fig.update_layout(height=600)
|
|
except Exception as e:
|
|
self.logger.exception(f"Failed to generate the bar chart graphic with error: {e}")
|
|
st.error(
|
|
f"An issue was encountered while generating the bar chart graphic. A detailed error message has been recorded in the logs. {self.app_config.get_errors_contact_string()}")
|
|
st.stop()
|
|
|
|
bar_fig_data = extract_figure_data(bar_fig)
|
|
if bar_fig_data.empty:
|
|
bar_fig_max_y = 0.0
|
|
else:
|
|
bar_fig_max_y = bar_fig_data.groupby('x')['y'].sum().max()
|
|
|
|
return {
|
|
'pie_fig': FigureWithMaxY(figure=pie_fig, max_y=0.0),
|
|
'bar_fig': FigureWithMaxY(figure=bar_fig, max_y=bar_fig_max_y),
|
|
'milestone_df':milestone_df
|
|
}
|
|
|
|
def render_figures(self, container: DeltaGenerator, output_data: Dict[str, Any]):
|
|
"""
|
|
Displays the generated visualizations and raw data in the UI.
|
|
|
|
Renders Plotly charts for documentation levels and attribution
|
|
sources, followed by an expander containing the underlying
|
|
pandas DataFrame for auditing.
|
|
|
|
:param container: The Streamlit container for output display.
|
|
:type container: DeltaGenerator
|
|
:param output_data: The result set from generate_figures.
|
|
:type output_data: Dict[str, Any]
|
|
"""
|
|
milestone_df:pd.DataFrame = output_data['milestone_df']
|
|
pie_fig = output_data['pie_fig'].get("figure")
|
|
bar_fig = output_data['bar_fig'].get("figure")
|
|
|
|
container.plotly_chart(pie_fig, key=self.get_widget_key("funding_milestone_pie_fig"))
|
|
container.plotly_chart(bar_fig, key=self.get_widget_key("funding_milestone_bar_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 Milestone Dataset")
|
|
dataset_expander.write(milestone_df)
|
|
|
|
def get_syncable_figure_keys(self) -> List[str]:
|
|
"""
|
|
Designates specific figures for axis synchronization.
|
|
|
|
Identifies the bar chart as a candidate for Y-axis scaling
|
|
normalization when multiple instances of this report are
|
|
viewed in a side-by-side comparison.
|
|
|
|
:return: List of keys ('bar_fig') available for synchronization.
|
|
:rtype: List[str]
|
|
"""
|
|
return ["bar_fig"] |