(
self,
output_dir: str,
mode: str,
model: str,
tdp: int,
rapl_dir: str = "/sys/class/powercap/intel-rapl/subsystem",
tracking_mode: str = "machine",
rapl_include_dram: bool = False,
rapl_prefer_psys: bool = False,
)
| 186 | @dataclass |
| 187 | class CPU(BaseHardware): |
| 188 | def __init__( |
| 189 | self, |
| 190 | output_dir: str, |
| 191 | mode: str, |
| 192 | model: str, |
| 193 | tdp: int, |
| 194 | rapl_dir: str = "/sys/class/powercap/intel-rapl/subsystem", |
| 195 | tracking_mode: str = "machine", |
| 196 | rapl_include_dram: bool = False, |
| 197 | rapl_prefer_psys: bool = False, |
| 198 | ): |
| 199 | assert tracking_mode in ["machine", "process"] |
| 200 | self._power_history: List[Power] = [] |
| 201 | self._output_dir = output_dir |
| 202 | self._mode = mode |
| 203 | self._model = model |
| 204 | self._tdp = tdp |
| 205 | self._is_generic_tdp = False |
| 206 | self._tracking_mode = tracking_mode |
| 207 | self._pid = psutil.Process().pid |
| 208 | self._cpu_count = count_cpus() |
| 209 | self._process = psutil.Process(self._pid) |
| 210 | # For process tracking: store last measurement time and CPU times |
| 211 | self._last_measurement_time: Optional[float] = None |
| 212 | self._last_cpu_times: Dict[int, float] = {} # pid -> total cpu time |
| 213 | |
| 214 | if self._mode == "intel_power_gadget": |
| 215 | self._intel_interface = IntelPowerGadget(self._output_dir) |
| 216 | elif self._mode == "intel_rapl": |
| 217 | self._intel_interface = IntelRAPL( |
| 218 | rapl_dir=rapl_dir, |
| 219 | rapl_include_dram=rapl_include_dram, |
| 220 | rapl_prefer_psys=rapl_prefer_psys, |
| 221 | ) |
| 222 | |
| 223 | def __repr__(self) -> str: |
| 224 | if self._mode != "constant": |
nothing calls this directly
no test coverage detected