Refactor classes to use create_folder function from utils.py

Author: sayalaruano

report/helpers/utils.py

@@ -6,23 +6,30 @@
 from typing import Type
 
 ## CHECKS
-def assert_path(filepath: str):
+def check_path(filepath: str) -> bool:
     """
-    Checks that fpath is a string and that it exists
+    Checks if the given file or folder path exists.
 
-    PARAMS
-    -----
-    - filepath (str): the filepath or folderpath
+    PARAMETERS
+    ---------
+    filepath : str
+        The file or folder path to check.
 
-    OUTPUTS
-    -----
-    - raises assertion error if filepath is not a string or doesn't exist
-    """
+    RETURNS
+    -------
+    bool
+        True if the path exists, False otherwise.
 
+    RAISES
+    ------
+    AssertionError
+        If the filepath is not a valid string.
+    """
+    # Assert that the filepath is a string
     assert isinstance(filepath, str), f"Filepath must be a string: {filepath}"
-    assert os.path.exists(
-        os.path.abspath(filepath)
-    ), f"Filepath does not exist: {os.path.abspath(filepath)}"
+
+    # Check if the path exists
+    return os.path.exists(os.path.abspath(filepath))
 
 
 def assert_enum_value(enum_class: Type[StrEnum], value: str, logger: logging.Logger) -> StrEnum:
@@ -55,6 +62,43 @@ def assert_enum_value(enum_class: Type[StrEnum], value: str, logger: logging.Log
         logger.error(f"Invalid value for {enum_class.__name__}: '{value}'. Expected values are: {expected_values}")
         raise ValueError(f"Invalid {enum_class.__name__}: {value}. Expected values are: {expected_values}")
 
+## FILE_SYSTEM
+def create_folder(directory_path: str, is_nested: bool = False) -> bool:
+    """
+    Create a folder. Optionally create nested directories if the specified path includes subdirectories.
+
+    Parameters
+    ----------
+    directory_path : str
+        The path of the directory to create.
+    is_nested : bool
+        A flag indicating whether to create nested directories (True uses os.makedirs, False uses os.mkdir).
+
+    Returns
+    -------
+    bool
+        True if the folder was created or False if it already existed.
+
+    Raises
+    ------
+    OSError
+        If there is an error creating the directory.
+    """
+    try:
+        if not check_path(directory_path):
+            if is_nested:
+                # Create the directory and any necessary parent directories
+                os.makedirs(directory_path, exist_ok=True)
+                return True
+            else:
+                # Create only the final directory (not nested)
+                os.mkdir(directory_path)
+                return True
+        else:
+            return False
+    except OSError as e:
+        raise OSError(f"Error creating directory '{directory_path}': {e}")
+
 ## LOGGING
 def get_basename(fname: None | str = None) -> str:
     """
@@ -81,7 +125,8 @@ def get_basename(fname: None | str = None) -> str:
     """
     if fname is not None:
         # PRECONDITION
-        assert_path(fname)
+        if not check_path(fname):
+            raise FileNotFoundError(f"The specified path does not exist: {fname}")
         # MAIN FUNCTIONS
         return os.path.splitext(os.path.basename(fname))[0]
     else:
@@ -165,7 +210,7 @@ def generate_log_filename(folder: str = "logs", suffix: str = "") -> str:
     log_filepath (str): the file path to the log file
     """
     # PRECONDITIONS
-    assert_path(folder)
+    create_folder(folder)
 
     # MAIN FUNCTION
     log_filename = get_time(incl_timezone=False) + "_" + suffix + ".log"

report/main.py

@@ -10,20 +10,20 @@
     report, report_metadata = yaml_manager.load_report_metadata('./report_metadata_micw2graph.yaml')
 
     # Create report view
-    doc_report = doc_reportview.QuartoReportView(report_metadata['report']['id'], 
-                                                report_metadata['report']['name'], 
-                                                report=report, 
-                                                report_type = assert_enum_value(ReportType, report_metadata['report']['report_type'], report.logger),
-                                                report_format = assert_enum_value(doc_reportview.ReportFormat, report_metadata['report']['report_format'], report.logger),
-                                                columns=None)
-    doc_report.generate_report()
-    doc_report.run_report()
+    # doc_report = doc_reportview.QuartoReportView(report_metadata['report']['id'], 
+    #                                             report_metadata['report']['name'], 
+    #                                             report=report, 
+    #                                             report_type = assert_enum_value(ReportType, report_metadata['report']['report_type'], report.logger),
+    #                                             report_format = assert_enum_value(doc_reportview.ReportFormat, report_metadata['report']['report_format'], report.logger),
+    #                                             columns=None)
+    # doc_report.generate_report()
+    # doc_report.run_report()
 
-    # st_report = st_reportview.StreamlitReportView(report_metadata['report']['id'], 
-    #                                              report_metadata['report']['name'], 
-    #                                              report=report, 
-    #                                              report_type = assert_enum_value(ReportType, report_metadata['report']['report_type'], report.logger),
-    #                                              columns=None)
-    # st_report.generate_report()
-    # st_report.run_report()
+    st_report = st_reportview.StreamlitReportView(report_metadata['report']['id'], 
+                                                 report_metadata['report']['name'], 
+                                                 report=report, 
+                                                 report_type = assert_enum_value(ReportType, report_metadata['report']['report_type'], report.logger),
+                                                 columns=None)
+    st_report.generate_report()
+    st_report.run_report()
 

report/quarto_reportview.py

@@ -3,6 +3,7 @@
 import report as r
 from enum import StrEnum, auto
 from typing import List, Optional
+from helpers.utils import create_folder
 
 class ReportFormat(StrEnum):
     HTML = auto()
@@ -26,21 +27,31 @@ def __init__(self, id: int, name: str, report: r.Report, report_type: r.ReportTy
         super().__init__(id, name=name, report=report, report_type = report_type, columns=columns)
         self.report_format = report_format
 
-    def generate_report(self, output_dir: str = BASE_DIR) -> None:
+    def generate_report(self, output_dir: str = BASE_DIR, static_dir: str = STATIC_FILES_DIR) -> None:
         """
         Generates the qmd file of the quarto report. It creates code for rendering each section and its subsections with all components.
 
         Parameters
         ----------
         output_dir : str, optional
             The folder where the generated report files will be saved (default is BASE_DIR).
+        static_dir : str, optional
+            The folder where the static files will be saved (default is STATIC_FILES_DIR).
         """
         self.report.logger.debug(f"Generating '{self.report_type}' report with '{self.report_format}' format in directory: '{output_dir}'")
 
-        # Create the output folder if it does not exist
-        if not os.path.exists(output_dir):
-            os.mkdir(output_dir)
-            self.report.logger.debug(f"Created output directory: {output_dir}")
+        # Create the output folder
+        if create_folder(output_dir):
+            self.report.logger.debug(f"Created output directory: '{output_dir}'")
+        else:
+            self.report.logger.debug(f"Output directory already existed: '{output_dir}'")
+
+        # Create the static folder
+        if create_folder(static_dir):
+            self.report.logger.info(f"Created output directory for static content: '{static_dir}'")
+        else:
+            self.report.logger.info(f"Output directory for static content already existed: '{static_dir}'")
+        
         try:
             # Create variable to check if the report is static or revealjs
             is_report_static = self.report_format in {ReportFormat.PDF, ReportFormat.DOCX, ReportFormat.ODT, ReportFormat.PPTX}
@@ -230,37 +241,31 @@ def _generate_subsection(self, subsection, is_report_static, is_report_revealjs)
         self.report.logger.info(f"Generated content and imports for subsection: '{subsection.name}'")
         return subsection_content, subsection_imports
 
-    def _generate_plot_content(self, plot, is_report_static, output_dir: str = STATIC_FILES_DIR) -> List[str]:
+    def _generate_plot_content(self, plot, is_report_static, static_dir: str = STATIC_FILES_DIR) -> List[str]:
         """
         Generate content for a plot component based on the report type.
 
         Parameters
         ----------
         plot : Plot
             The plot component to generate content for.
-        is_report_static : bool
-            A boolean indicating whether the report is static or interactive.
+        static_dir : str, optional
+            The folder where the static files will be saved (default is STATIC_FILES_DIR).
         
         Returns
         -------
         list : List[str]
             The list of content lines for the plot.
-        output_dir : str, optional
-            The folder where the static files will be saved (default is STATIC_FILES_DIR).
         """
-        # Create the output folder if it does not exist
-        if not os.path.exists(output_dir):
-            os.mkdir(output_dir)
-
         plot_content = []
         plot_content.append(f'### {plot.title}')
         if plot.plot_type == r.PlotType.INTERACTIVE:
             try:
                 # Define plot path
                 if is_report_static:
-                    static_plot_path  = os.path.join(output_dir, f"{plot.name.replace(' ', '_')}.png")
+                    static_plot_path  = os.path.join(static_dir, f"{plot.name.replace(' ', '_')}.png")
                 else:
-                    html_plot_file  = os.path.join(output_dir, f"{plot.name.replace(' ', '_')}.html")
+                    html_plot_file  = os.path.join(static_dir, f"{plot.name.replace(' ', '_')}.html")
 
                 if plot.int_visualization_tool == r.IntVisualizationTool.PLOTLY:
                     plot_content.append(self._generate_plot_code(plot))
@@ -447,7 +452,7 @@ def _generate_image_content(self, image_path: str, alt_text: str = "", width: in
         return f"""
 ![{alt_text}]({os.path.join('..', image_path)}){{ width={width}px height={height}px fig-align="center"}}\n"""
 
-    def _show_dataframe(self, dataframe, is_report_static, output_dir: str = STATIC_FILES_DIR) -> List[str]:
+    def _show_dataframe(self, dataframe, is_report_static, static_dir: str = STATIC_FILES_DIR) -> List[str]:
         """
         Appends either a static image or an interactive representation of a DataFrame to the content list.
 
@@ -457,18 +462,18 @@ def _show_dataframe(self, dataframe, is_report_static, output_dir: str = STATIC_
             The DataFrame object containing the data to display.
         is_report_static : bool
             Determines if the report is in a static format (e.g., PDF) or interactive (e.g., HTML).
+        static_dir : str, optional
+            The folder where the static files will be saved (default is STATIC_FILES_DIR).
         
         Returns
         -------
         list : List[str]
             The list of content lines for the DataFrame.
-        output_dir : str, optional
-            The folder where the static files will be saved (default is STATIC_FILES_DIR).
         """
         dataframe_content = []
         if is_report_static:
             # Generate path for the DataFrame image
-            df_image = os.path.join(output_dir, f"{dataframe.name.replace(' ', '_')}.png")
+            df_image = os.path.join(static_dir, f"{dataframe.name.replace(' ', '_')}.png")
             dataframe_content.append(f"dfi.export(df, '{os.path.join('..', df_image)}', max_rows=10, max_cols=5)\n```\n")
             # Use helper method to add centered image content
             dataframe_content.append(self._generate_image_content(df_image, dataframe.name))

report/streamlit_reportview.py

@@ -2,6 +2,7 @@
 import os
 import subprocess
 from typing import List, Optional
+from helpers.utils import create_folder
 
 class StreamlitReportView(r.WebAppReportView):
     """
@@ -28,9 +29,12 @@ def generate_report(self, output_dir: str = SECTIONS_DIR) -> None:
         self.report.logger.debug(f"Generating '{self.report_type}' report in directory: '{output_dir}'")
 
         # Create the output folder if it does not exist
-        if not os.path.exists(output_dir):
-            os.makedirs(output_dir, exist_ok=True)
+        #if not os.path.exists(output_dir):
+        #    os.makedirs(output_dir, exist_ok=True)
+        if create_folder(output_dir, is_nested=True):
             self.report.logger.debug(f"Created output directory: '{output_dir}'")
+        else:
+            self.report.logger.debug(f"Output directory already existed: '{output_dir}'")
 
         try:
             self.report.logger.debug("Processing app navigation code.")
@@ -51,13 +55,17 @@ def generate_report(self, output_dir: str = SECTIONS_DIR) -> None:
                 # Create a folder for each section
                 subsection_page_vars = []
                 section_name_var = section.name.replace(" ", "_")
-                if not os.path.exists(os.path.join(output_dir, section_name_var)):
-                    os.mkdir(os.path.join(output_dir, section_name_var))
-                    self.report.logger.debug(f"Created section directory: {section_name_var}")
+                section_dir_path = os.path.join(output_dir, section_name_var)
+                #if not os.path.exists(os.path.join(output_dir, section_name_var)):
+                # os.mkdir(os.path.join(output_dir, section_name_var))
+                if create_folder(section_dir_path):
+                    self.report.logger.debug(f"Created section directory: {section_dir_path}")
+                else:
+                    self.report.logger.debug(f"Section directory already existed: {section_dir_path}")
 
                 for subsection in section.subsections:
                     subsection_name_var = subsection.name.replace(" ", "_")
-                    subsection_file_path = os.path.join(section_name_var, section_name_var + "_" + subsection_name_var + ".py")
+                    subsection_file_path = os.path.join(section_name_var, subsection_name_var + ".py")
 
                     # Create a Page object for each subsection and add it to the home page content
                     report_manag_content.append(f"{subsection_name_var} = st.Page('{subsection_file_path}', title='{subsection.name}')")
@@ -142,11 +150,12 @@ def _generate_home_section(self, output_dir: str, report_manag_content: list) ->
         try:
             # Create folder for the home page
             home_dir_path = os.path.join(output_dir, "Home")
-            if not os.path.exists(home_dir_path):
-                os.mkdir(home_dir_path)
+            #if not os.path.exists(home_dir_path):
+            #os.mkdir(home_dir_path)
+            if create_folder(home_dir_path): 
                 self.report.logger.debug(f"Created home directory: {home_dir_path}")
             else:
-                self.report.logger.debug(f"Home directory already exists: {home_dir_path}")
+                self.report.logger.debug(f"Home directory already existed: {home_dir_path}")
 
             # Create the home page content
             home_content = []
@@ -192,7 +201,7 @@ def _generate_sections(self, output_dir: str) -> None:
                         self.report.logger.debug(f"Processing subsection: '{subsection.name} - {len(subsection.components)} component(s)'")
                         try:
                             # Create subsection file
-                            subsection_file_path = os.path.join(output_dir, section_name_var, section_name_var + "_" + subsection.name.replace(" ", "_") + ".py")
+                            subsection_file_path = os.path.join(output_dir, section_name_var, subsection.name.replace(" ", "_") + ".py")
 
                             # Generate content and imports for the subsection
                             subsection_content, subsection_imports = self._generate_subsection(subsection)
@@ -279,9 +288,12 @@ def _generate_plot_content(self, plot, output_dir: str = STATIC_FILES_DIR) -> Li
             The folder where the static files will be saved (default is STATIC_FILES_DIR).
         """
         # Create the output folder if it does not exist
-        if not os.path.exists(output_dir):
-            os.mkdir(output_dir)
+        #if not os.path.exists(output_dir):
+        #    os.mkdir(output_dir)
+        if create_folder(output_dir):
             self.report.logger.debug(f"Created output directory for static content: {output_dir}")
+        else:
+            self.report.logger.debug(f"Output directory for static content already existed: {output_dir}")
 
         plot_content = []
         plot_content.append(self._format_text(text=plot.title, type='header', level=4, color='#2b8cbe'))