1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164 | import datetime
import logging
import os
from codecarbon import EmissionsTracker
from pyJoules.device.device_factory import DeviceFactory
from pyJoules.device.nvidia_device import NvidiaGPUDomain
from pyJoules.device.rapl_device import RaplPackageDomain
from pyJoules.energy_meter import EnergyMeter
from pyJoules.handler.csv_handler import CSVHandler
from pyJoules.handler.pandas_handler import PandasHandler
from zeus.monitor import ZeusMonitor
class PowerTracker:
def __init__(
self,
power_tracker_library="codecarbon",
save_to_api=False,
output_dir="data/raw",
output_file="power_emissions.csv",
tracking_mode="machine",
):
self.save_to_api = save_to_api
self.output_dir = output_dir
self.output_file = output_file
self.tracking_mode = tracking_mode
self.output_dir = output_dir
self.output_file = output_file
self.power_tracker_library = power_tracker_library
self.csv_handler = None
self.tracker = self.init_tracker()
def init_tracker(self):
if self.power_tracker_library == "codecarbon":
tracker = self.init_codecarbon()
elif self.power_tracker_library == "pyjoules":
logging.info("Initializing pyJoules")
tracker = self.init_pyjoules()
elif self.power_tracker_library == "zeus":
tracker = self.init_zeus()
else:
raise ValueError("Invalid power tracker library")
return tracker
def init_codecarbon(self):
tracker = EmissionsTracker(
save_to_api=False,
output_dir=self.output_dir,
output_file=self.output_file,
tracking_mode="machine",
)
return tracker
def init_pyjoules(self):
self.csv_handler = CSVHandler(os.path.join(self.output_dir, self.output_file))
print(os.path.join(self.output_dir, self.output_file))
domains = [RaplPackageDomain(0), NvidiaGPUDomain(0)]
devices = DeviceFactory.create_devices()
tracker = EnergyMeter(devices)
return tracker
def init_zeus(self):
tracker = ZeusMonitor()
return tracker
def start(self):
print(f"Start time: {datetime.datetime.now()}")
if self.power_tracker_library == "codecarbon":
self.tracker.start()
elif self.power_tracker_library == "zeus":
self.tracker.begin_window("power")
elif self.power_tracker_library == "pyjoules":
print("Starting pyJoules")
self.tracker.start()
self.tracker.record(tag="record")
else:
raise ValueError("Invalid power tracker library")
def stop(self):
print(f"Stop time: {datetime.datetime.now()}")
if self.power_tracker_library == "codecarbon":
self.tracker.stop()
elif self.power_tracker_library == "zeus":
result = self.tracker.end_window("power")
time = result.time
energy = result.total_energy
# save to csv
with open(os.path.join(self.output_dir, self.output_file), "w") as f:
f.write(f"time,energy\n{time},{energy}")
elif self.power_tracker_library == "pyjoules":
self.tracker.stop()
trace = self.tracker.get_trace()
pandas_handler = PandasHandler()
pandas_handler.process(trace)
df = pandas_handler.get_dataframe()
df.to_csv(os.path.join(self.output_dir, self.output_file))
else:
raise ValueError("Invalid power tracker library")
def process_zeus_data(zeus_data):
"""Process Zeus measurement data.
Components:
- energy: Total energy consumed (J)
- time: Total time duration (s)
"""
zeus_energy = zeus_data["energy"][0] # J
zeus_time = zeus_data["time"][0] # s
zeus_power = zeus_energy / zeus_time
energy_kwh = zeus_energy / 3600 / 1e3 # Convert J to kWh
print("Zeus Energy Measurements:")
print(f"├── Duration: {zeus_time:.2f} seconds")
print(f"├── Average Power: {zeus_power:.2f} W")
print(f"└── Total Energy: {energy_kwh:.1e} kWh")
def process_pyjoules_data(pyjoules_data):
"""Process PyJoules measurement data.
Components:
- package_0: Complete CPU package (includes everything)
- core_0: CPU cores only
- uncore_0: Uncore components (cache, memory controller)
- nvidia_gpu_0: NVIDIA GPU (if present)
"""
measurements = {
"duration": pyjoules_data.iloc[1]["duration"], # seconds
"package": pyjoules_data.iloc[1]["package_0"] / 1e6, # µJ to J
"core": pyjoules_data.iloc[1]["core_0"] / 1e6, # µJ to J
"gpu": pyjoules_data.iloc[1]["nvidia_gpu_0"] / 1e6, # µJ to J
}
# Calculate power metrics
duration = measurements["duration"]
total_energy = measurements["package"] # Use package as it includes everything
average_power = total_energy / duration if duration > 0 else 0 # J/s = W
energy_kwh = total_energy / 3600 / 1e3 # Convert J to kWh
print("PyJoules Energy Measurements:")
print(f"├── Duration: {duration:.2f} seconds")
print(f"├── Package Power: {measurements['package']/duration:.6e} W")
print(f"├── Core Power: {measurements['core']/duration:.6e} W")
print(f"├── GPU Power: {measurements['gpu']/duration:.6e} W")
print(f"└── Total Energy: {energy_kwh:.1e} kWh")
return
def ampere_to_kWh(ampere, duration=1, volts=5.6, verbose=False):
# Calculate the average power
average_power = ampere.mean() * volts
if verbose:
print(f"Average Power: {average_power:.2f} W")
seconds_to_hour = 3600
kilo = 1000
# Calculate the total energy consumed
energy = average_power * duration / seconds_to_hour / kilo
return energy
|