Dynamic network carbon data energy with tests (#1257)

* Fix using correct SCI values for network energy calculation

* Add tests for dynamic network carbon data intensity

* Implement PR suggestions

Author: davidkopp

lib/phase_stats.py

@@ -8,7 +8,6 @@
 from decimal import Decimal
 from io import StringIO
 
-from lib.global_config import GlobalConfig
 from lib.db import DB
 from lib import error_helpers
 
@@ -68,8 +67,6 @@ def generate_csv_line(run_id, metric, detail_name, phase_name, value, value_type
     return f"{run_id},{metric},{detail_name},{phase_name},{round(value)},{value_type},{round(max_value) if max_value is not None else ''},{round(min_value) if min_value is not None else ''},{round(sampling_rate_avg) if sampling_rate_avg is not None else ''},{round(sampling_rate_max) if sampling_rate_max is not None else ''},{round(sampling_rate_95p) if sampling_rate_95p is not None else ''},{unit},NOW()\n"
 
 def build_and_store_phase_stats(run_id, sci=None):
-    config = GlobalConfig().config
-
     if not sci:
         sci = {}
 
@@ -263,17 +260,20 @@ def build_and_store_phase_stats(run_id, sci=None):
 
         # after going through detail metrics, create cumulated ones
         if network_bytes_total:
-            # build the network energy
-            # network via formula: https://www.green-coding.io/co2-formulas/
-            # pylint: disable=invalid-name
-            network_io_in_kWh = Decimal(sum(network_bytes_total)) / 1_000_000_000 * Decimal(config['sci']['N'])
-            network_io_in_uJ = network_io_in_kWh * 3_600_000_000_000
-            csv_buffer.write(generate_csv_line(run_id, 'network_energy_formula_global', '[FORMULA]', f"{idx:03}_{phase['name']}", network_io_in_uJ, 'TOTAL', None, None, None, None, None, 'uJ'))
-            # co2 calculations
-            network_io_carbon_in_ug = network_io_in_kWh * Decimal(config['sci']['I']) * 1_000_000
-            if '[' not in phase['name']: # only for runtime sub phases
-                software_carbon_intensity_global['network_io_carbon_in_ug'] = software_carbon_intensity_global.get('network_io_carbon_in_ug', 0) + network_io_carbon_in_ug
-            csv_buffer.write(generate_csv_line(run_id, 'network_carbon_formula_global', '[FORMULA]', f"{idx:03}_{phase['name']}", network_io_carbon_in_ug, 'TOTAL', None, None, None, None, None, 'ug'))
+            if sci.get('N', None) is not None and sci.get('I', None) is not None:
+                # build the network energy by using a formula: https://www.green-coding.io/co2-formulas/
+                # pylint: disable=invalid-name
+                network_io_in_kWh = Decimal(sum(network_bytes_total)) / 1_000_000_000 * Decimal(sci['N'])
+                network_io_in_uJ = network_io_in_kWh * 3_600_000_000_000
+                csv_buffer.write(generate_csv_line(run_id, 'network_energy_formula_global', '[FORMULA]', f"{idx:03}_{phase['name']}", network_io_in_uJ, 'TOTAL', None, None, None, None, None, 'uJ'))
+                # co2 calculations
+                network_io_carbon_in_ug = network_io_in_kWh * Decimal(sci['I']) * 1_000_000
+                if '[' not in phase['name']: # only for runtime sub phases
+                    software_carbon_intensity_global['network_io_carbon_in_ug'] = software_carbon_intensity_global.get('network_io_carbon_in_ug', 0) + network_io_carbon_in_ug
+                csv_buffer.write(generate_csv_line(run_id, 'network_carbon_formula_global', '[FORMULA]', f"{idx:03}_{phase['name']}", network_io_carbon_in_ug, 'TOTAL', None, None, None, None, None, 'ug'))
+            else:
+                error_helpers.log_error('Cannot calculate the total network energy consumption. SCI values I and N are missing in the config.', run_id=run_id)
+                network_io_carbon_in_ug = 0
         else:
             network_io_carbon_in_ug = 0
 
@@ -303,14 +303,14 @@ def build_and_store_phase_stats(run_id, sci=None):
                 csv_buffer.write(generate_csv_line(run_id, 'psu_energy_cgroup_container', detail_name, f"{idx:03}_{phase['name']}", surplus_energy_runtime * splitting_ratio, 'TOTAL', None, None, None, None, None, 'uJ'))
                 csv_buffer.write(generate_csv_line(run_id, 'psu_power_cgroup_container', detail_name, f"{idx:03}_{phase['name']}", surplus_power_runtime * splitting_ratio, 'TOTAL', None, None, None, None, None, 'mW'))
 
-    # TODO: refactor to be a metric provider. Than it can also be per phase
+    # TODO: refactor to be a metric provider. Than it can also be per phase # pylint: disable=fixme
     if software_carbon_intensity_global.get('machine_carbon_ug', None) is not None \
         and software_carbon_intensity_global.get('embodied_carbon_share_ug', None) is not None \
         and sci.get('R', 0) != 0 \
         and sci.get('R_d', None) is not None:
 
         csv_buffer.write(generate_csv_line(run_id, 'software_carbon_intensity_global', '[SYSTEM]', f"{runtime_phase_idx:03}_[RUNTIME]", (software_carbon_intensity_global['machine_carbon_ug'] + software_carbon_intensity_global['embodied_carbon_share_ug'] + software_carbon_intensity_global.get('network_io_carbon_in_ug', 0)) / Decimal(sci['R']), 'TOTAL', None, None, None, None, None, f"ugCO2e/{sci['R_d']}"))
-    # TODO End
+    # TODO End # pylint: disable=fixme
 
     csv_buffer.seek(0)  # Reset buffer position to the beginning
     DB().copy_from(

tests/lib/test_phase_stats.py

@@ -1,5 +1,7 @@
 import os
 import io
+import math
+from decimal import Decimal
 
 GMT_ROOT_DIR = os.path.dirname(os.path.abspath(__file__))+'/../../'
 
@@ -278,8 +280,124 @@ def test_phase_stats_single_network_procfs():
     assert data[14]['sampling_rate_95p'] == 100477, '95p sampling rate not in expected range'
     assert isinstance(data[14]['sampling_rate_95p'], int)
 
-
-def test_sci():
+def test_phase_stats_network_data():
+    run_id = Tests.insert_run()
+    Tests.import_network_io_cgroup_container(run_id)
+
+    test_sci_config = {
+        'N': 0.001,    # Network energy intensity (kWh/GB)
+        'I': 500,      # Carbon intensity (gCO2e/kWh)
+    }
+
+    build_and_store_phase_stats(run_id, sci=test_sci_config)
+
+    # Network energy data
+    network_energy_data = DB().fetch_all(
+        'SELECT metric, detail_name, unit, value, type, phase FROM phase_stats WHERE phase = %s AND metric = %s',
+        params=('004_[RUNTIME]', 'network_energy_formula_global'), fetch_mode='dict'
+    )
+
+    assert len(network_energy_data) == 1, f"Expected 1 network energy formula entry, got {len(network_energy_data)}"
+
+    network_energy_entry = network_energy_data[0]
+    assert network_energy_entry['metric'] == 'network_energy_formula_global'
+    assert network_energy_entry['detail_name'] == '[FORMULA]'
+    assert network_energy_entry['unit'] == 'uJ'
+    assert network_energy_entry['type'] == 'TOTAL'
+    assert network_energy_entry['phase'] == '004_[RUNTIME]'
+
+    network_totals = DB().fetch_all(
+        'SELECT detail_name, value FROM phase_stats WHERE phase = %s AND metric = %s',
+        params=('004_[RUNTIME]', 'network_total_cgroup_container'), fetch_mode='dict'
+    )
+    total_network_bytes = sum(row['value'] for row in network_totals)
+    expected_network_energy_kwh = Decimal(total_network_bytes) / 1_000_000_000 * Decimal(test_sci_config['N'])
+    expected_network_energy_uj = expected_network_energy_kwh * 3_600_000_000_000
+    assert math.isclose(network_energy_entry['value'], expected_network_energy_uj, rel_tol=1e-5), f"Expected network energy: {expected_network_energy_uj}, got: {network_energy_entry['value']}"
+
+    # Network carbon data
+    network_carbon_data = DB().fetch_all(
+        'SELECT metric, detail_name, unit, value, type FROM phase_stats WHERE phase = %s AND metric = %s',
+        params=('004_[RUNTIME]', 'network_carbon_formula_global'), fetch_mode='dict'
+    )
+
+    assert len(network_carbon_data) == 1, "Expected 1 network carbon formula entry"
+
+    network_carbon_entry = network_carbon_data[0]
+    expected_network_carbon_ug = expected_network_energy_kwh * Decimal(test_sci_config['I']) * 1_000_000
+
+    assert network_carbon_entry['metric'] == 'network_carbon_formula_global'
+    assert network_carbon_entry['detail_name'] == '[FORMULA]'
+    assert network_carbon_entry['unit'] == 'ug'
+    assert network_carbon_entry['type'] == 'TOTAL'
+    assert math.isclose(network_carbon_entry['value'], expected_network_carbon_ug, rel_tol=1e-5), f"Expected network carbon: {expected_network_carbon_ug}, got: {network_carbon_entry['value']}"
+
+def test_sci_calculation():
+    run_id = Tests.insert_run()
+    Tests.import_machine_energy(run_id)  # Machine energy component
+    Tests.import_network_io_cgroup_container(run_id)  # Network component (custom N parameter)
+
+    # Define comprehensive SCI configuration with all required parameters
+    test_sci_config = {
+        'N': 0.001,    # Network energy intensity (kWh/GB)
+        'I': 500,      # Carbon intensity (gCO2e/kWh)
+        'EL': 4,       # Expected lifespan (years)
+        'TE': 300000,  # Total embodied emissions (gCO2e)
+        'RS': 1,       # Resource share (100%)
+        'R': 10,       # Functional unit count (10 runs)
+        'R_d': 'test runs'  # Functional unit description
+    }
+
+    build_and_store_phase_stats(run_id, sci=test_sci_config)
+
+    # Verify all SCI components are calculated and stored correctly
+
+    # 1. Machine carbon from energy consumption
+    machine_carbon_data = DB().fetch_all(
+        'SELECT metric, value, unit FROM phase_stats WHERE phase = %s AND metric = %s',
+        params=('004_[RUNTIME]', 'psu_carbon_ac_mcp_machine'), fetch_mode='dict'
+    )
+    assert len(machine_carbon_data) == 1, "Machine carbon should be calculated"
+    machine_carbon_ug = machine_carbon_data[0]['value']
+
+    # 2. Embodied carbon calculation
+    embodied_carbon_data = DB().fetch_all(
+        'SELECT metric, value, unit FROM phase_stats WHERE phase = %s AND metric = %s',
+        params=('004_[RUNTIME]', 'embodied_carbon_share_machine'), fetch_mode='dict'
+    )
+    assert len(embodied_carbon_data) == 1, "Embodied carbon should be calculated"
+    embodied_carbon_ug = embodied_carbon_data[0]['value']
+
+    # 3. Network carbon calculation
+    network_carbon_data = DB().fetch_all(
+        'SELECT metric, value, unit FROM phase_stats WHERE phase = %s AND metric = %s',
+        params=('004_[RUNTIME]', 'network_carbon_formula_global'), fetch_mode='dict'
+    )
+    assert len(network_carbon_data) == 1, "Network carbon should be calculated"
+    network_carbon_ug = network_carbon_data[0]['value']
+
+    # 4. Final SCI calculation verification
+    sci_data = DB().fetch_all(
+        'SELECT value, unit FROM phase_stats WHERE phase = %s AND metric = %s',
+        params=('004_[RUNTIME]', 'software_carbon_intensity_global'), fetch_mode='dict'
+    )
+    assert len(sci_data) == 1, "SCI should be calculated for the whole run"
+    sci_entry = sci_data[0]
+
+    # Verify SCI unit format includes functional unit description - fail if other unit occurs
+    expected_unit = f"ugCO2e/{test_sci_config['R_d']}"
+    assert sci_entry['unit'] == expected_unit, \
+        f"Test fails: Unexpected unit detected. Expected: {expected_unit}, got: {sci_entry['unit']}. This test is designed to fail when incorrect units are present."
+
+    # Verify SCI value matches expected value: (machine_carbon + embodied_carbon + network_carbon) / R
+    expected_sci_value = (machine_carbon_ug + embodied_carbon_ug + network_carbon_ug) / Decimal(test_sci_config['R'])
+    assert math.isclose(abs(sci_entry['value']), expected_sci_value, rel_tol=1e-5), f"SCI calculation should be correct. Expected: {expected_sci_value}, got: {sci_entry['value']}"
+
+    # Verify SCI value is reasonable (positive and within expected range)
+    assert sci_entry['value'] > 0, "SCI should be positive"
+    assert sci_entry['value'] < 1000000, "SCI should be reasonable (less than 1M ugCO2e per functional unit)"
+
+def test_sci_run():
     runner = ScenarioRunner(uri=GMT_ROOT_DIR, uri_type='folder', filename='tests/data/usage_scenarios/stress_sci.yml', skip_system_checks=True, dev_cache_build=True, dev_no_sleeps=True, dev_no_metrics=False, dev_no_phase_stats=False)
 
     out = io.StringIO()
@@ -289,7 +407,6 @@ def test_sci():
 
     data = DB().fetch_all("SELECT value, unit FROM phase_stats WHERE phase = %s AND run_id = %s AND metric = 'software_carbon_intensity_global' ", params=('004_[RUNTIME]', run_id), fetch_mode='dict')
 
-
     assert len(data) == 1
     assert 50 < data[0]['value'] < 150
     assert data[0]['unit'] == 'ugCO2e/Cool run'

tests/test_functions.py

@@ -11,6 +11,7 @@
 from metric_providers.psu.energy.ac.mcp.machine.provider import PsuEnergyAcMcpMachineProvider
 from metric_providers.cpu.energy.rapl.msr.component.provider import CpuEnergyRaplMsrComponentProvider
 from metric_providers.network.io.procfs.system.provider import NetworkIoProcfsSystemProvider
+from metric_providers.network.io.cgroup.container.provider import NetworkIoCgroupContainerProvider
 
 CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
 
@@ -127,6 +128,17 @@ def import_network_io_procfs(run_id):
 
     return df
 
+def import_network_io_cgroup_container(run_id):
+
+    obj = NetworkIoCgroupContainerProvider(99, skip_check=True)
+
+    obj._filename = os.path.join(CURRENT_DIR, 'data/metrics/network_io_cgroup_container.log')
+    df = obj.read_metrics()
+
+    metric_importer.import_measurements(df, 'network_io_cgroup_container', run_id)
+
+    return df
+
 def import_cpu_energy(run_id):
 
     obj = CpuEnergyRaplMsrComponentProvider(99, skip_check=True)