Merging changes to use ga_core package from Dev to Main

.github/workflows/python-app.yml

@@ -0,0 +1,45 @@
+# This workflow will install Python dependencies, run tests and lint with a single version of Python
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: Python application
+
+on:
+  push:
+    branches: [ "main", "dev" ]
+  pull_request:
+    branches: [ "main", "dev" ]
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python 3.13
+      uses: actions/setup-python@v5
+      with:
+        python-version: "3.13"
+        cache: "pip"
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install flake8 pytest pytest-cov
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+
+    - name: Lint with flake8
+      run: |
+        # stop the build if there are Python syntax errors or undefined names
+        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
+        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+
+    - name: Test with pytest
+      env:
+          PYTHONPATH: .
+      run: |
+        # Generates pytest-coverage.txt used by the comment action
+        pytest --doctest-modules --junitxml=junit/test-results.xml --cov-report=term-missing --cov=. | tee pytest-coverage.txt

.gitignore

@@ -1,7 +1,6 @@
 # Project specific
 .idea/
 clustersData/
-testData/
 error_logs_archived/
 support_files/
 frontend/templates/plotly*
@@ -145,4 +144,8 @@ dmypy.json
 .pytype/
 
 # Cython debug symbols
-cython_debug/
+cython_debug/
+.vscode/launch.json
+.gitignore
+.vscode/settings.json
+.DS_Store

__init__.py

@@ -2,8 +2,9 @@
 import argparse
 import datetime
 import os
+import sys
 
-from backend import main_backend
+from backend import export_debug_logs, main_backend
 from frontend import main_frontend
 
 def create_arguments():
@@ -59,13 +60,14 @@ def create_arguments():
                         )
     group1.add_argument('--reportBugHere', action='store_true',
                         help='Similar to --reportBug, but exports the output to your home folder.')
+
     group2 = parser.add_mutually_exclusive_group()
     group2.add_argument('--useCustomLogs', type=str, default='',
                         help='This bypasses the workload manager, and enables you to input a custom log file of your jobs. \
                                  This is mostly meant for debugging, but can be useful in some situations. '
                              'An example of the expected file can be found at `example_files/example_sacctOutput_raw.txt`.')
     # Arguments for debugging only (not visible to users)
-    # To ue arbitrary folder for the infrastructure information
+    # To use arbitrary folder for the infrastructure information
     parser.add_argument('--useOtherInfrastuctureInfo', type=str, default='', help=argparse.SUPPRESS)
     # Uses mock aggregated usage data, for offline debugging
     group2.add_argument('--use_mock_agg_data', action='store_true', help=argparse.SUPPRESS)
@@ -120,7 +122,7 @@ def all(self, args):
     else:
         args.path_infrastucture_info = 'data'
 
-    ## Organise the unique output directory (used for output report and logs export for debugging)
+    ## Organise the unique output directory (used for output report)
     ## creating a uniquely named subdirectory in whatever
     # Decide if an output directory is needed at all
     if (args.output in ['html']) | args.reportBug | args.reportBugHere:
@@ -144,6 +146,16 @@ def all(self, args):
     else:
         args.filterWD = None
 
+    if args.reportBug | args.reportBugHere:
+            print("\n(!) Debugging mode activated. This will export raw logs for debugging.\n")
+            if args.useCustomLogs != '':
+                print("\n(!) --reportBug and --reportBugHere are ignored when --useCustomLogs is present\n")
+            else:
+                # Extract and save debug logs
+                export_debug_logs(args)
+                print("\n(!) Exiting after exporting debug logs.\n")
+                sys.exit(0)
+
     ### Validate input
     validate_args().all(args)
 

backend/__init__.py

@@ -1,125 +1,13 @@
 
 import os
+import sys
 import yaml
-import pandas as pd
-import numpy as np
-
-from backend.helpers import check_empty_results, simulate_mock_jobs
-from backend.slurm_extract import WorkloadManager
+import ga_core
 
+from backend import helpers
 # print("Working dir1: ", os.getcwd()) # DEBUGONLY
 
-class GA_tools():
-
-    def __init__(self, cluster_info, fParams):
-        self.cluster_info = cluster_info
-        self.fParams = fParams
-
-    def calculate_energies(self, row):
-        '''
-        Calculate the energy usaged based on the job's paramaters
-        :param row: [pd.Series] one row of usage statistics, corresponding to one job
-        :return: [pd.Series] the same statistics with the energies added
-        '''
-        ### CPU and GPU
-        partition_info = self.cluster_info['partitions'][row.PartitionX]
-        if row.PartitionTypeX == 'CPU':
-            TDP2use4CPU = partition_info['TDP']
-            TDP2use4GPU = 0
-        else:
-            TDP2use4CPU = partition_info['TDP_CPU']
-            TDP2use4GPU = partition_info['TDP']
-
-        row['energy_CPUs'] = row.TotalCPUtime2useX.total_seconds() / 3600 * TDP2use4CPU / 1000  # in kWh
-
-        row['energy_GPUs'] = row.TotalGPUtime2useX.total_seconds() / 3600 * TDP2use4GPU / 1000  # in kWh
-
-        ### memory
-        for suffix, memory2use in zip(['','_memoryNeededOnly'], [row.ReqMemX,row.NeededMemX]):
-            row[f'energy_memory{suffix}'] = row.WallclockTimeX.total_seconds()/3600 * memory2use * self.fParams['power_memory_perGB'] /1000 # in kWh
-            row[f'energy{suffix}'] = (row.energy_CPUs +  row.energy_GPUs + row[f'energy_memory{suffix}']) * self.cluster_info['PUE'] # in kWh
-
-        return row
-
-    def calculate_carbonFootprint(self, df, col_energy):
-        return df[col_energy] * self.cluster_info['CI']
-
-
-def extract_data(args, cluster_info):
-
-    if args.use_mock_agg_data: # DEBUGONLY
-
-        if args.reportBug | args.reportBugHere:
-            print("\n(!) --reportBug and --reportBugHere are ignored when --useCustomLogs is present\n")
-
-        # df2 = simulate_mock_jobs()
-        # df2.to_pickle("testData/df_agg_X_mockMultiUsers_1.pkl")
-
-        # foo = 'testData/df_agg_test_3.pkl'
-        foo = 'testData/df_agg_X_1.pkl'
-        print(f"Overriding df_agg with `{foo}`")
-        return pd.read_pickle(foo)
-
-
-    ### Pull usage statistics from the workload manager
-    WM = WorkloadManager(args, cluster_info)
-    WM.pull_logs()
-
-    ### Log the output for debugging
-    if args.reportBug | args.reportBugHere:
-        if args.reportBug:
-            # Create an error_logs subfolder in the output dir
-            errorLogsDir = os.path.join(args.outputDir2use['path'], 'error_logs')
-            os.makedirs(errorLogsDir)
-            log_path = os.path.join(errorLogsDir, f'sacctOutput.csv')
-        else:
-            # i.e. args.reportBugHere is True
-            log_path = f"{args.userCWD}/sacctOutput_{args.outputDir2use['timestamp']}.csv"
-
-        with open(log_path, 'wb') as f:
-            f.write(WM.logs_raw)
-        print(f"\nSLURM statistics logged for debuging: {log_path}\n")
-
-    ### Turn usage logs into DataFrame
-    WM.convert2dataframe()
-    check_empty_results(WM.logs_df, args)
-
-    # And clean
-    WM.clean_logs_df()
-    # Check if there are any jobs during the period from this directory and with these jobIDs
-    check_empty_results(WM.df_agg, args)
-
-    # Check that there is only one user's data
-    if len(set(WM.df_agg_X.UserX)) > 1:
-        raise ValueError(f"More than one user's logs was included: {set(WM.df_agg_X.UserX)}")
-
-    # WM.df_agg_X.to_pickle("testData/df_agg_X_1.pkl") # DEBUGONLY used to test different steps offline
-
-    return WM.df_agg_X
-
-def enrich_data(df, fParams, GA):
-
-    ### energy
-    df = df.apply(GA.calculate_energies, axis=1)
-
-    df['energy_failedJobs'] = np.where(df.StateX == 0, df.energy, 0)
-
-    ### carbon footprint
-    for suffix in ['', '_memoryNeededOnly', '_failedJobs']:
-        df[f'carbonFootprint{suffix}'] = GA.calculate_carbonFootprint(df, f'energy{suffix}')
-        # Context metrics (part 1)
-        df[f'treeMonths{suffix}'] = df[f'carbonFootprint{suffix}'] / fParams['tree_month']
-        df[f'cost{suffix}'] = df[f'energy{suffix}'] * fParams['electricity_cost'] # TODO use realtime electricity costs
-
-    ### Context metrics (part 2)
-    df['driving'] = df.carbonFootprint / fParams['passengerCar_EU_perkm']
-    df['flying_NY_SF'] = df.carbonFootprint / fParams['flight_NY_SF']
-    df['flying_PAR_LON'] = df.carbonFootprint / fParams['flight_PAR_LON']
-    df['flying_NYC_MEL'] = df.carbonFootprint / fParams['flight_NYC_MEL']
-
-    return df
-
-def summarise_data(df, args):
+def summarise_data(df):
     agg_functions_from_raw = {
         'n_jobs': ('UserX', 'count'),
         'first_job_period': ('SubmitDatetimeX', 'min'),
@@ -190,7 +78,7 @@ def agg_jobs(data, agg_names=None):
     df_userdaily = agg_jobs(df, ['SubmitDate'])
     df_overallStats = agg_jobs(df_userdaily)
     dict_overallStats = df_overallStats.iloc[0, :].to_dict()
-    userID = df.UserX[0]
+    userID = df.UserX.iloc[0]
 
     output = {
         "userDaily": df_userdaily,
@@ -205,17 +93,34 @@ def agg_jobs(data, agg_names=None):
 
     return output
 
+def prepare_config(args):
+    """
+    Prepare the configuration for the GA core, based on the command line arguments.
+    :param args: [argparse.Namespace] the command line arguments
+    :return: [dict] the configuration for the GA core
+    """
+    ga_config = {
+        "useCustomLogs": args.useCustomLogs,
+        "startDay": args.startDay,
+        "endDay": args.endDay,
+        "filterWD": args.filterWD,
+        "filterJobIDs": args.filterJobIDs,
+        "filterAccount": args.filterAccount
+        }
+
+    # TODO: Need to be implemented in a better manner, perhaps by importing a model from ga_core
+    optional_args = ["userCWD", "customSuccessStates"] 
+    for arg in optional_args:
+        if hasattr(args, arg) and getattr(args, arg):
+            ga_config[arg] = getattr(args, arg)
 
-def main_backend(args):
-    '''
-
-    :param args:
-    :return:
-    '''
     ### Load cluster specific info
     with open(os.path.join(args.path_infrastucture_info, 'cluster_info.yaml'), "r") as stream:
         try:
             cluster_info = yaml.safe_load(stream)
+            if cluster_info.get('workload_manager', '') == '':
+                cluster_info['workload_manager'] = 'slurm'  # default to slurm if not specified
+
         except yaml.YAMLError as exc:
             print(exc)
 
@@ -226,33 +131,83 @@ def main_backend(args):
         except yaml.YAMLError as exc:
             print(exc)
 
-    GA = GA_tools(cluster_info, fParams)
+    return ga_config, cluster_info, fParams
+
+def main_backend(args):
+    '''
+    Loads configurations including cluster information and fixed parameters.
+    Calls HPCDataProcessor.extract and HPCDataProcessor.enrich functions to produce enriched logs.
+    Finally, it summarises the data.
+
+    :param args: [argparse.Namespace] contains the settings
+    :return: [dict] contains the summarised data
+    '''
+    ga_config, cluster_info, fParams = prepare_config(args)
+    logs_raw = None
+
+    if ga_config.get('useCustomLogs', '') != '':
+        # Pick raw logs from file
+        logs_raw = helpers.read_file_bytes(ga_config["useCustomLogs"])
+        print(f'Overriding logs_raw with: {ga_config["useCustomLogs"]}\n')     
 
-    df = extract_data(args, cluster_info=cluster_info)
-    df2 = enrich_data(df, fParams=fParams, GA=GA)
-    summary_stats = summarise_data(df2, args=args)
+    dataprocessor = ga_core.HPCDataProcessor(ga_config, cluster_info, fParams, all_users_access = False)
+    extracted_logs = dataprocessor.extract_data(logs_raw)
+
+    enriched_logs = dataprocessor.enrich_data(extracted_logs)
+    summary_stats = summarise_data(enriched_logs)
 
     return summary_stats
 
+def export_debug_logs(args) -> None:
+    """
+    Exports raw logs to a CSV file for debugging.
+
+    :param args: [argparse.Namespace] contains the settings
+    """
+    ga_config, cluster_info, fParams = prepare_config(args)
+
+    if args.reportBug:
+        # Create an error_logs subfolder in the output dir
+        errorLogsDir = os.path.join(args.outputDir2use['path'], 'error_logs')
+        os.makedirs(errorLogsDir)
+        log_path = os.path.join(errorLogsDir, f'extracted_output.txt')
+    else:
+        # i.e. args.reportBugHere is True
+        log_path = f"{args.userCWD}/extracted_output_{args.outputDir2use['timestamp']}.txt"
+
+    try:
+        match cluster_info.get('workload_manager', '').lower():
+            case 'slurm':
+                extracted_raw_logs = ga_core.SacctClient.pull_logs_by_time(startDay=ga_config['startDay'], endDay=ga_config['endDay'], all_users=False)
+                with open(log_path, 'wb') as f:
+                    f.write(extracted_raw_logs)
+                print(f"\nSLURM statistics logged for debugging: {log_path}\n")
+            case _:
+                raise ValueError(f"Unsupported workload manager: {cluster_info['workload_manager']}")
+
+    except IOError as e:
+        print(f"\n[Debug logs] Failed to write debug logs to {log_path}: {e}\n")
+
+    except Exception as e:
+            print(f"[Debug logs] Failed to extract logs: {e}")
+            sys.exit(1)
+
 if __name__ == "__main__":
 
-    #### This is used for testing only ####
+    #### This is used for testing/DEBUG only ####
 
     from collections import namedtuple
     argStruct = namedtuple('argStruct',
-                           'startDay endDay use_mock_agg_data useCustomLogs customSuccessStates filterWD filterJobIDs filterAccount reportBug reportBugHere path_infrastucture_info')
+                           'startDay endDay useCustomLogs customSuccessStates filterWD filterJobIDs filterAccount path_infrastucture_info')
     args = argStruct(
         startDay='2022-01-01',
         endDay='2023-06-30',
-        useCustomLogs=None,
-        use_mock_agg_data=True,
+        useCustomLogs='',
         customSuccessStates='',
         filterWD=None,
         filterJobIDs='all',
         filterAccount=None,
-        reportBug=False,
-        reportBugHere=False,
-        path_infrastucture_info="clustersData/CSD3",
+        path_infrastucture_info="data/",
     )
 
     main_backend(args)