emi

zeus.device.cpu.emi

Windows EMI (Energy Meter Interface) CPU energy monitoring.

EMI (Energy Meter Interface): EMI is a Windows interface introduced in Windows 10 that allows applications to read energy consumption data from hardware energy meters. It provides access to RAPL (Running Average Power Limit) counters on Intel processors via a standardized IOCTL interface.
Energy Meter Device: An EMI device represents one energy metering unit. On Intel systems, a single EMI device typically exposes multiple channels corresponding to different power domains (e.g., package, DRAM, PP0, PP1) for each CPU socket.
Channel: Each EMI device exposes one or more named channels. Channel names follow the pattern RAPL_Package{N}_{DOMAIN} where N is the socket index and DOMAIN is the power domain (e.g., PKG, DRAM, PP0, PP1).

See: https://learn.microsoft.com/en-us/windows-hardware/drivers/powermeter/energy-meter-interface

ZeusEMINotSupportedError

Bases: ZeusBaseCPUError

Raised when EMI is not available on this system.

Source code in zeus/device/cpu/emi.py

class ZeusEMINotSupportedError(ZeusBaseCPUError):
    """Raised when EMI is not available on this system."""

    def __init__(self, message: str) -> None:
        """Initialize Zeus Exception."""
        super().__init__(message)

init

__init__(message)

Source code in zeus/device/cpu/emi.py

def __init__(self, message: str) -> None:
    """Initialize Zeus Exception."""
    super().__init__(message)

ZeusEMIInitError

Bases: ZeusBaseCPUError

Raised when an EMI device cannot be opened or queried.

Source code in zeus/device/cpu/emi.py

class ZeusEMIInitError(ZeusBaseCPUError):
    """Raised when an EMI device cannot be opened or queried."""

    def __init__(self, message: str) -> None:
        """Initialize Zeus Exception."""
        super().__init__(message)

init

__init__(message)

Source code in zeus/device/cpu/emi.py

def __init__(self, message: str) -> None:
    """Initialize Zeus Exception."""
    super().__init__(message)

_EMIChannel

Metadata for a single EMI channel.

Source code in zeus/device/cpu/emi.py

class _EMIChannel:
    """Metadata for a single EMI channel."""

    __slots__ = ("index", "name", "unit")

    def __init__(self, index: int, name: str, unit: int) -> None:
        self.index = index
        self.name = name
        self.unit = unit

EMIFile

Manages an open Windows EMI device handle and reads energy data from it.

Each EMIFile corresponds to one EMI device interface (one device path). It reads energy values in picowatt-hours and converts them to millijoules.

Attributes:

Name	Type	Description
`path`	`str`	The Windows device interface path.
`version`	`int`	The EMI interface version reported by the device (1 or 2).
`channels`	`list[_EMIChannel]`	Metadata for each channel on this device.

Source code in zeus/device/cpu/emi.py

class EMIFile:
    """Manages an open Windows EMI device handle and reads energy data from it.

    Each ``EMIFile`` corresponds to one EMI device interface (one device path).
    It reads energy values in picowatt-hours and converts them to millijoules.

    Attributes:
        path (str): The Windows device interface path.
        version (int): The EMI interface version reported by the device (1 or 2).
        channels (list[_EMIChannel]): Metadata for each channel on this device.
    """

    def __init__(self, path: str) -> None:
        r"""Open the EMI device and read its metadata.

        Args:
            path: Windows device interface path (e.g. ``\\?\acpi#...``).

        Raises:
            ZeusEMIInitError: If the device cannot be opened or its metadata cannot be read.
        """
        self.path = path

        handle = _kernel32.CreateFileW(
            path,
            _GENERIC_READ,
            _FILE_SHARE_READ,
            None,
            _OPEN_EXISTING,
            0,
            None,
        )
        if handle == _INVALID_HANDLE_VALUE:
            err = get_last_error()
            raise ZeusEMIInitError(f"Failed to open EMI device '{path}' (Windows error {err}).")
        # Store the raw integer handle value so that __del__ can safely check
        # ``isinstance(self._handle, int)`` and avoid calling CloseHandle on
        # mocked handles during testing.
        self._handle: int = int(handle)

        try:
            self.version, self.channels = self._read_metadata()
        except ZeusEMIInitError:
            _kernel32.CloseHandle(self._handle)
            raise

    def _read_metadata(self) -> tuple[int, list[_EMIChannel]]:
        """Query the device for its version and channel metadata.

        Returns:
            A (version, channels) tuple.

        Raises:
            ZeusEMIInitError: On any IOCTL failure.
        """
        # Version
        ver_bytes = _ioctl(ctypes.c_void_p(self._handle), _IOCTL_EMI_GET_VERSION, 2)
        if ver_bytes is None or len(ver_bytes) < 2:
            raise ZeusEMIInitError(f"IOCTL_EMI_GET_VERSION failed for '{self.path}'.")
        version = int.from_bytes(ver_bytes[:2], "little")
        if version not in (_EMI_VERSION_V1, _EMI_VERSION_V2):
            raise ZeusEMIInitError(f"Unsupported EMI version {version} for '{self.path}'.")

        # Metadata size
        ms_bytes = _ioctl(ctypes.c_void_p(self._handle), _IOCTL_EMI_GET_METADATA_SIZE, 4)
        if ms_bytes is None or len(ms_bytes) < 4:
            raise ZeusEMIInitError(f"IOCTL_EMI_GET_METADATA_SIZE failed for '{self.path}'.")
        meta_size = int.from_bytes(ms_bytes[:4], "little")
        if meta_size == 0:
            raise ZeusEMIInitError(f"EMI device '{self.path}' reported zero metadata size.")

        # Metadata payload
        meta_bytes = _ioctl(ctypes.c_void_p(self._handle), _IOCTL_EMI_GET_METADATA, meta_size)
        if meta_bytes is None:
            raise ZeusEMIInitError(f"IOCTL_EMI_GET_METADATA failed for '{self.path}'.")

        channels = self._parse_channels(version, meta_bytes)
        return version, channels

    @staticmethod
    def _parse_channels(version: int, raw: bytes) -> list[_EMIChannel]:
        """Parse channel metadata from the raw metadata buffer.

        EMI_METADATA_V1 layout (offsets in bytes):
            MeasurementUnit  UINT   4
            HardwareOEM      WCHAR[16]  32
            HardwareModel    WCHAR[16]  32
            HardwareRevision USHORT 2
            MeteredHardwareNameSize USHORT 2
            MeteredHardwareName WCHAR[] variable

        EMI_METADATA_V2 layout (offsets in bytes):
            HardwareOEM      WCHAR[16]  32   @ 0
            HardwareModel    WCHAR[16]  32   @ 32
            HardwareRevision USHORT     2    @ 64
            ChannelCount     USHORT     2    @ 66
            Channels[]                       @ 68

        Each EMI_CHANNEL_V2:
            MeasurementUnit  UINT   4
            ChannelNameSize  USHORT 2   (bytes, including null terminator)
            ChannelName      WCHAR[] ChannelNameSize bytes
        """
        channels: list[_EMIChannel] = []

        if version == _EMI_VERSION_V2:
            # _EMI_NAME_MAX = 16 WCHARs = 32 bytes each for OEM and Model.
            _oem_size = _EMI_NAME_MAX * 2
            _model_size = _EMI_NAME_MAX * 2
            header_size = _oem_size + _model_size + 4  # OEM, Model, Revision, ChannelCount
            if len(raw) < header_size:
                raise ZeusEMIInitError(f"EMI V2 metadata is too short ({len(raw)} bytes).")
            channel_count = int.from_bytes(raw[header_size - 2 : header_size], "little")
            offset = header_size

            for i in range(channel_count):
                if offset + 6 > len(raw):
                    raise ZeusEMIInitError(
                        f"EMI V2 metadata is truncated: expected {channel_count} channels, "
                        f"but the buffer ({len(raw)} bytes) ends inside channel {i}."
                    )
                unit = int.from_bytes(raw[offset : offset + 4], "little")
                name_size = int.from_bytes(raw[offset + 4 : offset + 6], "little")
                if offset + 6 + name_size > len(raw):
                    raise ZeusEMIInitError(
                        f"EMI V2 metadata is truncated: channel {i}'s name extends past "
                        f"the end of the buffer ({len(raw)} bytes)."
                    )
                name = (
                    raw[offset + 6 : offset + 6 + name_size].decode("utf-16-le").rstrip("\x00") if name_size > 0 else ""
                )
                channels.append(_EMIChannel(index=i, name=name, unit=unit))
                offset += 4 + 2 + name_size

        elif version == _EMI_VERSION_V1:
            _oem_size = _EMI_NAME_MAX * 2
            _model_size = _EMI_NAME_MAX * 2
            header_size = 4 + _oem_size + _model_size + 4  # unit, OEM, Model, revision, name size
            if len(raw) < header_size:
                raise ZeusEMIInitError(f"EMI V1 metadata is too short ({len(raw)} bytes).")
            # Single channel: the whole device is one domain.
            unit = int.from_bytes(raw[0:4], "little")
            name_size = int.from_bytes(
                raw[4 + _oem_size + _model_size + 2 : 4 + _oem_size + _model_size + 4],
                "little",
            )
            if len(raw) < header_size + name_size:
                raise ZeusEMIInitError(f"EMI V1 metadata is truncated for the channel name ({len(raw)} bytes).")
            name = (
                raw[4 + _oem_size + _model_size + 4 : 4 + _oem_size + _model_size + 4 + name_size]
                .decode("utf-16-le")
                .rstrip("\x00")
                if name_size > 0
                else "EMI_V1"
            )
            channels.append(_EMIChannel(index=0, name=name, unit=unit))

        return channels

    def read(self, channel_index: int) -> float:
        """Read the accumulated energy for the given channel.

        Args:
            channel_index: Zero-based index of the channel within this device.

        Returns:
            The accumulated energy in millijoules.

        Raises:
            ZeusEMIInitError: If the channel is invalid, the channel's measurement
                unit is not picowatt-hours, or the IOCTL call fails.
        """
        if not 0 <= channel_index < len(self.channels):
            raise ZeusEMIInitError(
                f"Channel index {channel_index} is out of range for '{self.path}' ({len(self.channels)} channels)."
            )
        channel = self.channels[channel_index]
        if channel.unit != _EMI_MEASUREMENT_UNIT_PICOWATT_HOURS:
            raise ZeusEMIInitError(
                f"EMI channel '{channel.name}' reports measurement unit {channel.unit}, "
                "not picowatt-hours (0); refusing to convert to millijoules."
            )
        meas_size = len(self.channels) * 16  # 16 bytes per EMI_CHANNEL_MEASUREMENT_DATA
        meas_bytes = _ioctl(ctypes.c_void_p(self._handle), _IOCTL_EMI_GET_MEASUREMENT, meas_size)
        if meas_bytes is None:
            err = get_last_error()
            raise ZeusEMIInitError(f"IOCTL_EMI_GET_MEASUREMENT failed for '{self.path}' (Windows error {err}).")
        # EMI_MEASUREMENT_DATA_V2: ChannelData[ChannelCount]
        # Each EMI_CHANNEL_MEASUREMENT_DATA: AbsoluteEnergy (8 bytes) + AbsoluteTime (8 bytes)
        offset = channel_index * 16
        if len(meas_bytes) < offset + 8:
            raise ZeusEMIInitError(
                f"Measurement data from '{self.path}' is too short "
                f"({len(meas_bytes)} bytes, expected at least {offset + 8})."
            )
        energy_pwh = int.from_bytes(meas_bytes[offset : offset + 8], "little")
        return energy_pwh * _PICOWATT_HOURS_TO_MILLIJOULES

    def __del__(self) -> None:
        """Close the device handle.

        During interpreter shutdown module globals may already be cleared, so we
        guard against ``_kernel32``/``ctypes`` being ``None`` and swallow any
        error rather than raising from a finalizer.
        """
        if not _WINDOWS or _kernel32 is None or ctypes is None:
            return
        handle = getattr(self, "_handle", 0)
        if isinstance(handle, int) and handle and handle != _INVALID_HANDLE_VALUE:
            with contextlib.suppress(Exception):
                _kernel32.CloseHandle(ctypes.c_void_p(handle))
            self._handle = 0

init

__init__(path)

Parameters:

Name	Type	Description	Default
`path`	`str`	Windows device interface path (e.g. `\\?\acpi#...`).	required

Raises:

Type	Description
`ZeusEMIInitError`	If the device cannot be opened or its metadata cannot be read.

Source code in zeus/device/cpu/emi.py

def __init__(self, path: str) -> None:
    r"""Open the EMI device and read its metadata.

    Args:
        path: Windows device interface path (e.g. ``\\?\acpi#...``).

    Raises:
        ZeusEMIInitError: If the device cannot be opened or its metadata cannot be read.
    """
    self.path = path

    handle = _kernel32.CreateFileW(
        path,
        _GENERIC_READ,
        _FILE_SHARE_READ,
        None,
        _OPEN_EXISTING,
        0,
        None,
    )
    if handle == _INVALID_HANDLE_VALUE:
        err = get_last_error()
        raise ZeusEMIInitError(f"Failed to open EMI device '{path}' (Windows error {err}).")
    # Store the raw integer handle value so that __del__ can safely check
    # ``isinstance(self._handle, int)`` and avoid calling CloseHandle on
    # mocked handles during testing.
    self._handle: int = int(handle)

    try:
        self.version, self.channels = self._read_metadata()
    except ZeusEMIInitError:
        _kernel32.CloseHandle(self._handle)
        raise

_read_metadata

_read_metadata()

Query the device for its version and channel metadata.

Returns:

Type	Description
`tuple[int, list[_EMIChannel]]`	A (version, channels) tuple.

Raises:

Type	Description
`ZeusEMIInitError`	On any IOCTL failure.

Source code in zeus/device/cpu/emi.py

def _read_metadata(self) -> tuple[int, list[_EMIChannel]]:
    """Query the device for its version and channel metadata.

    Returns:
        A (version, channels) tuple.

    Raises:
        ZeusEMIInitError: On any IOCTL failure.
    """
    # Version
    ver_bytes = _ioctl(ctypes.c_void_p(self._handle), _IOCTL_EMI_GET_VERSION, 2)
    if ver_bytes is None or len(ver_bytes) < 2:
        raise ZeusEMIInitError(f"IOCTL_EMI_GET_VERSION failed for '{self.path}'.")
    version = int.from_bytes(ver_bytes[:2], "little")
    if version not in (_EMI_VERSION_V1, _EMI_VERSION_V2):
        raise ZeusEMIInitError(f"Unsupported EMI version {version} for '{self.path}'.")

    # Metadata size
    ms_bytes = _ioctl(ctypes.c_void_p(self._handle), _IOCTL_EMI_GET_METADATA_SIZE, 4)
    if ms_bytes is None or len(ms_bytes) < 4:
        raise ZeusEMIInitError(f"IOCTL_EMI_GET_METADATA_SIZE failed for '{self.path}'.")
    meta_size = int.from_bytes(ms_bytes[:4], "little")
    if meta_size == 0:
        raise ZeusEMIInitError(f"EMI device '{self.path}' reported zero metadata size.")

    # Metadata payload
    meta_bytes = _ioctl(ctypes.c_void_p(self._handle), _IOCTL_EMI_GET_METADATA, meta_size)
    if meta_bytes is None:
        raise ZeusEMIInitError(f"IOCTL_EMI_GET_METADATA failed for '{self.path}'.")

    channels = self._parse_channels(version, meta_bytes)
    return version, channels

_parse_channels `staticmethod`

_parse_channels(version, raw)

Parse channel metadata from the raw metadata buffer.

EMI_METADATA_V1 layout (offsets in bytes): MeasurementUnit UINT 4 HardwareOEM WCHAR[16] 32 HardwareModel WCHAR[16] 32 HardwareRevision USHORT 2 MeteredHardwareNameSize USHORT 2 MeteredHardwareName WCHAR[] variable

EMI_METADATA_V2 layout (offsets in bytes): HardwareOEM WCHAR[16] 32 @ 0 HardwareModel WCHAR[16] 32 @ 32 HardwareRevision USHORT 2 @ 64 ChannelCount USHORT 2 @ 66 Channels[] @ 68

Each EMI_CHANNEL_V2

MeasurementUnit UINT 4 ChannelNameSize USHORT 2 (bytes, including null terminator) ChannelName WCHAR[] ChannelNameSize bytes

Source code in zeus/device/cpu/emi.py

@staticmethod
def _parse_channels(version: int, raw: bytes) -> list[_EMIChannel]:
    """Parse channel metadata from the raw metadata buffer.

    EMI_METADATA_V1 layout (offsets in bytes):
        MeasurementUnit  UINT   4
        HardwareOEM      WCHAR[16]  32
        HardwareModel    WCHAR[16]  32
        HardwareRevision USHORT 2
        MeteredHardwareNameSize USHORT 2
        MeteredHardwareName WCHAR[] variable

    EMI_METADATA_V2 layout (offsets in bytes):
        HardwareOEM      WCHAR[16]  32   @ 0
        HardwareModel    WCHAR[16]  32   @ 32
        HardwareRevision USHORT     2    @ 64
        ChannelCount     USHORT     2    @ 66
        Channels[]                       @ 68

    Each EMI_CHANNEL_V2:
        MeasurementUnit  UINT   4
        ChannelNameSize  USHORT 2   (bytes, including null terminator)
        ChannelName      WCHAR[] ChannelNameSize bytes
    """
    channels: list[_EMIChannel] = []

    if version == _EMI_VERSION_V2:
        # _EMI_NAME_MAX = 16 WCHARs = 32 bytes each for OEM and Model.
        _oem_size = _EMI_NAME_MAX * 2
        _model_size = _EMI_NAME_MAX * 2
        header_size = _oem_size + _model_size + 4  # OEM, Model, Revision, ChannelCount
        if len(raw) < header_size:
            raise ZeusEMIInitError(f"EMI V2 metadata is too short ({len(raw)} bytes).")
        channel_count = int.from_bytes(raw[header_size - 2 : header_size], "little")
        offset = header_size

        for i in range(channel_count):
            if offset + 6 > len(raw):
                raise ZeusEMIInitError(
                    f"EMI V2 metadata is truncated: expected {channel_count} channels, "
                    f"but the buffer ({len(raw)} bytes) ends inside channel {i}."
                )
            unit = int.from_bytes(raw[offset : offset + 4], "little")
            name_size = int.from_bytes(raw[offset + 4 : offset + 6], "little")
            if offset + 6 + name_size > len(raw):
                raise ZeusEMIInitError(
                    f"EMI V2 metadata is truncated: channel {i}'s name extends past "
                    f"the end of the buffer ({len(raw)} bytes)."
                )
            name = (
                raw[offset + 6 : offset + 6 + name_size].decode("utf-16-le").rstrip("\x00") if name_size > 0 else ""
            )
            channels.append(_EMIChannel(index=i, name=name, unit=unit))
            offset += 4 + 2 + name_size

    elif version == _EMI_VERSION_V1:
        _oem_size = _EMI_NAME_MAX * 2
        _model_size = _EMI_NAME_MAX * 2
        header_size = 4 + _oem_size + _model_size + 4  # unit, OEM, Model, revision, name size
        if len(raw) < header_size:
            raise ZeusEMIInitError(f"EMI V1 metadata is too short ({len(raw)} bytes).")
        # Single channel: the whole device is one domain.
        unit = int.from_bytes(raw[0:4], "little")
        name_size = int.from_bytes(
            raw[4 + _oem_size + _model_size + 2 : 4 + _oem_size + _model_size + 4],
            "little",
        )
        if len(raw) < header_size + name_size:
            raise ZeusEMIInitError(f"EMI V1 metadata is truncated for the channel name ({len(raw)} bytes).")
        name = (
            raw[4 + _oem_size + _model_size + 4 : 4 + _oem_size + _model_size + 4 + name_size]
            .decode("utf-16-le")
            .rstrip("\x00")
            if name_size > 0
            else "EMI_V1"
        )
        channels.append(_EMIChannel(index=0, name=name, unit=unit))

    return channels

read

read(channel_index)

Read the accumulated energy for the given channel.

Parameters:

Name	Type	Description	Default
`channel_index`	`int`	Zero-based index of the channel within this device.	required

Returns:

Type	Description
`float`	The accumulated energy in millijoules.

Raises:

Type	Description
`ZeusEMIInitError`	If the channel is invalid, the channel's measurement unit is not picowatt-hours, or the IOCTL call fails.

Source code in zeus/device/cpu/emi.py

def read(self, channel_index: int) -> float:
    """Read the accumulated energy for the given channel.

    Args:
        channel_index: Zero-based index of the channel within this device.

    Returns:
        The accumulated energy in millijoules.

    Raises:
        ZeusEMIInitError: If the channel is invalid, the channel's measurement
            unit is not picowatt-hours, or the IOCTL call fails.
    """
    if not 0 <= channel_index < len(self.channels):
        raise ZeusEMIInitError(
            f"Channel index {channel_index} is out of range for '{self.path}' ({len(self.channels)} channels)."
        )
    channel = self.channels[channel_index]
    if channel.unit != _EMI_MEASUREMENT_UNIT_PICOWATT_HOURS:
        raise ZeusEMIInitError(
            f"EMI channel '{channel.name}' reports measurement unit {channel.unit}, "
            "not picowatt-hours (0); refusing to convert to millijoules."
        )
    meas_size = len(self.channels) * 16  # 16 bytes per EMI_CHANNEL_MEASUREMENT_DATA
    meas_bytes = _ioctl(ctypes.c_void_p(self._handle), _IOCTL_EMI_GET_MEASUREMENT, meas_size)
    if meas_bytes is None:
        err = get_last_error()
        raise ZeusEMIInitError(f"IOCTL_EMI_GET_MEASUREMENT failed for '{self.path}' (Windows error {err}).")
    # EMI_MEASUREMENT_DATA_V2: ChannelData[ChannelCount]
    # Each EMI_CHANNEL_MEASUREMENT_DATA: AbsoluteEnergy (8 bytes) + AbsoluteTime (8 bytes)
    offset = channel_index * 16
    if len(meas_bytes) < offset + 8:
        raise ZeusEMIInitError(
            f"Measurement data from '{self.path}' is too short "
            f"({len(meas_bytes)} bytes, expected at least {offset + 8})."
        )
    energy_pwh = int.from_bytes(meas_bytes[offset : offset + 8], "little")
    return energy_pwh * _PICOWATT_HOURS_TO_MILLIJOULES

del

__del__()

Close the device handle.

During interpreter shutdown module globals may already be cleared, so we guard against _kernel32/ctypes being None and swallow any error rather than raising from a finalizer.

Source code in zeus/device/cpu/emi.py

def __del__(self) -> None:
    """Close the device handle.

    During interpreter shutdown module globals may already be cleared, so we
    guard against ``_kernel32``/``ctypes`` being ``None`` and swallow any
    error rather than raising from a finalizer.
    """
    if not _WINDOWS or _kernel32 is None or ctypes is None:
        return
    handle = getattr(self, "_handle", 0)
    if isinstance(handle, int) and handle and handle != _INVALID_HANDLE_VALUE:
        with contextlib.suppress(Exception):
            _kernel32.CloseHandle(ctypes.c_void_p(handle))
        self._handle = 0

EMICPU

Bases: CPU

Reads energy for a single Intel CPU package via the Windows EMI interface.

Attributes:

Name	Type	Description
`cpu_index`	`int`	Zero-based package (socket) index.

Source code in zeus/device/cpu/emi.py

class EMICPU(cpu_common.CPU):
    """Reads energy for a single Intel CPU package via the Windows EMI interface.

    Attributes:
        cpu_index (int): Zero-based package (socket) index.
    """

    def __init__(
        self,
        cpu_index: int,
        emi_file: EMIFile,
        pkg_channel_index: int,
        dram_channel_index: int | None,
    ) -> None:
        """Initialize the EMICPU.

        Args:
            cpu_index: Zero-based CPU package (socket) index.
            emi_file: The :class:`EMIFile` that owns the device handle.
            pkg_channel_index: Index of the PKG (package) energy channel in ``emi_file``.
            dram_channel_index: Index of the DRAM energy channel, or ``None`` if unavailable.
        """
        super().__init__(cpu_index)
        self._emi_file = emi_file
        self._pkg_channel_index = pkg_channel_index
        self._dram_channel_index = dram_channel_index

    def get_total_energy_consumption(self) -> CpuDramMeasurement:
        """Return the total accumulated energy for this CPU package. Units: mJ."""
        cpu_mj = self._emi_file.read(self._pkg_channel_index)
        dram_mj: float | None = None
        if self._dram_channel_index is not None:
            dram_mj = self._emi_file.read(self._dram_channel_index)
        return CpuDramMeasurement(cpu_mj=cpu_mj, dram_mj=dram_mj)

    def supports_get_dram_energy_consumption(self) -> bool:
        """Return ``True`` if DRAM energy data is available for this package."""
        return self._dram_channel_index is not None

init

__init__(cpu_index, emi_file, pkg_channel_index, dram_channel_index)

Parameters:

Name	Type	Description	Default
`cpu_index`	`int`	Zero-based CPU package (socket) index.	required
`emi_file`	`EMIFile`	The :class:`EMIFile` that owns the device handle.	required
`pkg_channel_index`	`int`	Index of the PKG (package) energy channel in `emi_file`.	required
`dram_channel_index`	`int \| None`	Index of the DRAM energy channel, or `None` if unavailable.	required

Source code in zeus/device/cpu/emi.py

def __init__(
    self,
    cpu_index: int,
    emi_file: EMIFile,
    pkg_channel_index: int,
    dram_channel_index: int | None,
) -> None:
    """Initialize the EMICPU.

    Args:
        cpu_index: Zero-based CPU package (socket) index.
        emi_file: The :class:`EMIFile` that owns the device handle.
        pkg_channel_index: Index of the PKG (package) energy channel in ``emi_file``.
        dram_channel_index: Index of the DRAM energy channel, or ``None`` if unavailable.
    """
    super().__init__(cpu_index)
    self._emi_file = emi_file
    self._pkg_channel_index = pkg_channel_index
    self._dram_channel_index = dram_channel_index

get_total_energy_consumption

get_total_energy_consumption()

Return the total accumulated energy for this CPU package. Units: mJ.

Source code in zeus/device/cpu/emi.py

def get_total_energy_consumption(self) -> CpuDramMeasurement:
    """Return the total accumulated energy for this CPU package. Units: mJ."""
    cpu_mj = self._emi_file.read(self._pkg_channel_index)
    dram_mj: float | None = None
    if self._dram_channel_index is not None:
        dram_mj = self._emi_file.read(self._dram_channel_index)
    return CpuDramMeasurement(cpu_mj=cpu_mj, dram_mj=dram_mj)

supports_get_dram_energy_consumption

supports_get_dram_energy_consumption()

Return True if DRAM energy data is available for this package.

Source code in zeus/device/cpu/emi.py

def supports_get_dram_energy_consumption(self) -> bool:
    """Return ``True`` if DRAM energy data is available for this package."""
    return self._dram_channel_index is not None

EMICPUs

Bases: CPUs

Manages all Intel CPU packages accessible via the Windows EMI interface.

Each detected RAPL_Package{N}_PKG EMI channel maps to one :class:EMICPU object at index N.

Source code in zeus/device/cpu/emi.py

class EMICPUs(cpu_common.CPUs):
    """Manages all Intel CPU packages accessible via the Windows EMI interface.

    Each detected ``RAPL_Package{N}_PKG`` EMI channel maps to one
    :class:`EMICPU` object at index N.
    """

    def __init__(self) -> None:
        """Discover and initialise all EMI CPU objects.

        Raises:
            ZeusEMINotSupportedError: If EMI is unavailable on this system.
            ZeusEMIInitError: If device metadata cannot be queried.
        """
        if not emi_is_available():
            raise ZeusEMINotSupportedError(
                "No EMI energy meter devices exposing RAPL package channels were found on this system."
            )
        self._cpus: list[EMICPU] = []
        self._init_cpus()

    def _init_cpus(self) -> None:
        """Build the list of :class:`EMICPU` objects from all EMI devices."""
        paths = _get_emi_device_paths()

        # pkg_index → (EMIFile, pkg_channel_idx, dram_channel_idx | None)
        packages: dict[int, tuple[EMIFile, int, int | None]] = {}

        for path in paths:
            try:
                emi_file = EMIFile(path)
            except ZeusEMIInitError as err:
                logger.warning("Skipping EMI device '%s': %s", path, err)
                continue

            # Locate PKG and DRAM channels for each package on this device.
            # Channels with a non-picowatt-hour unit cannot be converted to
            # millijoules, so they are excluded rather than misconverted.
            pkg_channels: dict[int, int] = {}
            dram_channels: dict[int, int] = {}
            for ch in emi_file.channels:
                m = _PKG_CHANNEL_RE.match(ch.name)
                target = pkg_channels
                if m is None:
                    m = _DRAM_CHANNEL_RE.match(ch.name)
                    target = dram_channels
                if m is None:
                    continue
                if ch.unit != _EMI_MEASUREMENT_UNIT_PICOWATT_HOURS:
                    logger.warning(
                        "Ignoring EMI channel '%s' on '%s': unexpected measurement unit %d (expected picowatt-hours).",
                        ch.name,
                        path,
                        ch.unit,
                    )
                    continue
                target[int(m.group(1))] = ch.index

            for pkg_num, pkg_idx in pkg_channels.items():
                dram_idx = dram_channels.get(pkg_num)
                packages[pkg_num] = (emi_file, pkg_idx, dram_idx)

        if not packages:
            raise ZeusEMINotSupportedError("EMI devices were found but no RAPL_Package PKG channels were detected.")

        for pkg_num in sorted(packages):
            emi_file, pkg_idx, dram_idx = packages[pkg_num]
            self._cpus.append(
                EMICPU(
                    cpu_index=pkg_num,
                    emi_file=emi_file,
                    pkg_channel_index=pkg_idx,
                    dram_channel_index=dram_idx,
                )
            )
            logger.info(
                "Initialized EMI CPU %d: PKG channel=%d, DRAM channel=%s",
                pkg_num,
                pkg_idx,
                dram_idx,
            )

    @property
    def cpus(self) -> Sequence[EMICPU]:
        """Return the list of :class:`EMICPU` objects."""
        return self._cpus

    def __del__(self) -> None:
        """Clean up resources."""
        pass

cpus `property`

cpus

Return the list of :class:EMICPU objects.

init

__init__()

Raises:

Type	Description
`ZeusEMINotSupportedError`	If EMI is unavailable on this system.
`ZeusEMIInitError`	If device metadata cannot be queried.

Source code in zeus/device/cpu/emi.py

def __init__(self) -> None:
    """Discover and initialise all EMI CPU objects.

    Raises:
        ZeusEMINotSupportedError: If EMI is unavailable on this system.
        ZeusEMIInitError: If device metadata cannot be queried.
    """
    if not emi_is_available():
        raise ZeusEMINotSupportedError(
            "No EMI energy meter devices exposing RAPL package channels were found on this system."
        )
    self._cpus: list[EMICPU] = []
    self._init_cpus()

_init_cpus

_init_cpus()

Build the list of :class:EMICPU objects from all EMI devices.

Source code in zeus/device/cpu/emi.py

def _init_cpus(self) -> None:
    """Build the list of :class:`EMICPU` objects from all EMI devices."""
    paths = _get_emi_device_paths()

    # pkg_index → (EMIFile, pkg_channel_idx, dram_channel_idx | None)
    packages: dict[int, tuple[EMIFile, int, int | None]] = {}

    for path in paths:
        try:
            emi_file = EMIFile(path)
        except ZeusEMIInitError as err:
            logger.warning("Skipping EMI device '%s': %s", path, err)
            continue

        # Locate PKG and DRAM channels for each package on this device.
        # Channels with a non-picowatt-hour unit cannot be converted to
        # millijoules, so they are excluded rather than misconverted.
        pkg_channels: dict[int, int] = {}
        dram_channels: dict[int, int] = {}
        for ch in emi_file.channels:
            m = _PKG_CHANNEL_RE.match(ch.name)
            target = pkg_channels
            if m is None:
                m = _DRAM_CHANNEL_RE.match(ch.name)
                target = dram_channels
            if m is None:
                continue
            if ch.unit != _EMI_MEASUREMENT_UNIT_PICOWATT_HOURS:
                logger.warning(
                    "Ignoring EMI channel '%s' on '%s': unexpected measurement unit %d (expected picowatt-hours).",
                    ch.name,
                    path,
                    ch.unit,
                )
                continue
            target[int(m.group(1))] = ch.index

        for pkg_num, pkg_idx in pkg_channels.items():
            dram_idx = dram_channels.get(pkg_num)
            packages[pkg_num] = (emi_file, pkg_idx, dram_idx)

    if not packages:
        raise ZeusEMINotSupportedError("EMI devices were found but no RAPL_Package PKG channels were detected.")

    for pkg_num in sorted(packages):
        emi_file, pkg_idx, dram_idx = packages[pkg_num]
        self._cpus.append(
            EMICPU(
                cpu_index=pkg_num,
                emi_file=emi_file,
                pkg_channel_index=pkg_idx,
                dram_channel_index=dram_idx,
            )
        )
        logger.info(
            "Initialized EMI CPU %d: PKG channel=%d, DRAM channel=%s",
            pkg_num,
            pkg_idx,
            dram_idx,
        )

del

__del__()

Clean up resources.

Source code in zeus/device/cpu/emi.py

def __del__(self) -> None:
    """Clean up resources."""
    pass

_build_emi_guid

_build_emi_guid()

Build the EMI device interface GUID structure.

Source code in zeus/device/cpu/emi.py

def _build_emi_guid() -> "_Guid":
    """Build the EMI device interface GUID structure."""
    guid = _Guid()
    guid.Data1 = _EMI_GUID_DATA1
    guid.Data2 = _EMI_GUID_DATA2
    guid.Data3 = _EMI_GUID_DATA3
    for i, byte in enumerate(_EMI_GUID_DATA4):
        guid.Data4[i] = byte
    return guid

_get_emi_device_paths

_get_emi_device_paths()

Enumerate all EMI-compliant device interface paths on this Windows system.

Returns an empty list if no devices are found or if not running on Windows.

Source code in zeus/device/cpu/emi.py

def _get_emi_device_paths() -> list[str]:
    """Enumerate all EMI-compliant device interface paths on this Windows system.

    Returns an empty list if no devices are found or if not running on Windows.
    """
    guid = _build_emi_guid()

    hdev = _setupapi.SetupDiGetClassDevsW(
        ctypes.byref(guid),
        None,
        None,
        _DIGCF_PRESENT | _DIGCF_DEVICEINTERFACE,
    )
    if hdev == _INVALID_HANDLE_VALUE:
        logger.debug("SetupDiGetClassDevsW returned INVALID_HANDLE_VALUE for EMI GUID.")
        return []

    paths: list[str] = []
    try:
        index = 0
        while True:
            iface = _SpDeviceInterfaceData()
            iface.cbSize = ctypes.sizeof(_SpDeviceInterfaceData)
            if not _setupapi.SetupDiEnumDeviceInterfaces(
                ctypes.c_void_p(hdev),
                None,
                ctypes.byref(guid),
                index,
                ctypes.byref(iface),
            ):
                break

            # First call: obtain required buffer size.
            required = wintypes.DWORD(0)
            _setupapi.SetupDiGetDeviceInterfaceDetailW(
                ctypes.c_void_p(hdev),
                ctypes.byref(iface),
                None,
                0,
                ctypes.byref(required),
                None,
            )
            if required.value == 0:
                index += 1
                continue

            # Second call: fill the detail buffer.
            # SP_DEVICE_INTERFACE_DETAIL_DATA_W layout:
            #   DWORD cbSize  (4 bytes)
            #   WCHAR DevicePath[ANYSIZE_ARRAY]  (variable)
            detail_buf = ctypes.create_string_buffer(required.value)
            ctypes.cast(detail_buf, ctypes.POINTER(wintypes.DWORD))[0] = _DETAIL_DATA_CBSIZE
            if _setupapi.SetupDiGetDeviceInterfaceDetailW(
                ctypes.c_void_p(hdev),
                ctypes.byref(iface),
                detail_buf,
                required,
                None,
                None,
            ):
                # DevicePath starts immediately after the DWORD cbSize field.
                path = ctypes.wstring_at(ctypes.addressof(detail_buf) + 4)
                paths.append(path)
                logger.debug("Found EMI device: %s", path)

            index += 1
    finally:
        _setupapi.SetupDiDestroyDeviceInfoList(ctypes.c_void_p(hdev))

    return paths

_ioctl

_ioctl(handle, code, out_size)

Send a buffered IOCTL with no input buffer and return the output bytes.

Returns None if the call fails.

Source code in zeus/device/cpu/emi.py

def _ioctl(handle: "ctypes.c_void_p", code: int, out_size: int) -> bytes | None:
    """Send a buffered IOCTL with no input buffer and return the output bytes.

    Returns ``None`` if the call fails.
    """
    buf = ctypes.create_string_buffer(out_size)
    bytes_returned = wintypes.DWORD(0)
    ok = _kernel32.DeviceIoControl(
        handle,
        code,
        None,
        0,
        buf,
        out_size,
        ctypes.byref(bytes_returned),
        None,
    )
    if not ok:
        return None
    return bytes(buf)[: bytes_returned.value]

_find_package_of_processor

_find_package_of_processor(raw, group, number, ptr_size)

Find the index of the processor package containing a logical processor.

Parses the SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX records returned by GetLogicalProcessorInformationEx with RelationProcessorPackage and returns the zero-based index of the package whose group affinity contains the logical processor identified by processor group group and within-group number number, or None if no package matches.

Record layout: DWORD Relationship, DWORD Size, then PROCESSOR_RELATIONSHIP (BYTE Flags, BYTE EfficiencyClass, BYTE Reserved[20], WORD GroupCount, GROUP_AFFINITY GroupMask[GroupCount]). Each GROUP_AFFINITY is a pointer-sized KAFFINITY Mask, WORD Group, and WORD Reserved[3].

Raises:

Type	Description
`ValueError`	If the buffer is malformed.

Source code in zeus/device/cpu/emi.py

def _find_package_of_processor(raw: bytes, group: int, number: int, ptr_size: int) -> int | None:
    """Find the index of the processor package containing a logical processor.

    Parses the SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX records returned by
    `GetLogicalProcessorInformationEx` with `RelationProcessorPackage` and
    returns the zero-based index of the package whose group affinity contains
    the logical processor identified by processor group `group` and
    within-group number `number`, or `None` if no package matches.

    Record layout: DWORD Relationship, DWORD Size, then PROCESSOR_RELATIONSHIP
    (BYTE Flags, BYTE EfficiencyClass, BYTE Reserved[20], WORD GroupCount,
    GROUP_AFFINITY GroupMask[GroupCount]). Each GROUP_AFFINITY is a
    pointer-sized KAFFINITY Mask, WORD Group, and WORD Reserved[3].

    Raises:
        ValueError: If the buffer is malformed.
    """
    group_affinity_size = ptr_size + 2 + 6  # Mask + Group + Reserved[3]
    offset = 0
    package_index = 0
    while offset < len(raw):
        if offset + 8 > len(raw):
            raise ValueError("Truncated SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX record header.")
        record_size = int.from_bytes(raw[offset + 4 : offset + 8], "little")
        # 8-byte header + 24 bytes of PROCESSOR_RELATIONSHIP before the group array.
        if record_size < 8 + 24 or offset + record_size > len(raw):
            raise ValueError("Malformed SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX record size.")
        group_count = int.from_bytes(raw[offset + 8 + 22 : offset + 8 + 24], "little")
        for g in range(group_count):
            ga_offset = offset + 8 + 24 + g * group_affinity_size
            if ga_offset + group_affinity_size > offset + record_size:
                raise ValueError("GROUP_AFFINITY entries extend past their record.")
            mask = int.from_bytes(raw[ga_offset : ga_offset + ptr_size], "little")
            ga_group = int.from_bytes(raw[ga_offset + ptr_size : ga_offset + ptr_size + 2], "little")
            if ga_group == group and (mask >> number) & 1:
                return package_index
        package_index += 1
        offset += record_size
    return None

get_current_emi_cpu_index

get_current_emi_cpu_index()

Return the EMI CPU index of the package the calling thread is running on.

This is the EMI counterpart of get_current_rapl_zone_id in the RAPL module: the returned index can be passed as a cpu_indices entry to only measure the CPU package the current thread is running on.

The current logical processor (processor group and within-group number) is resolved with GetCurrentProcessorNumberEx and mapped to a package index with GetLogicalProcessorInformationEx. Package records are assumed to be enumerated in the same order as the EMI RAPL_Package{N} channel numbering.

Note

The scheduler can migrate threads across packages at any time. To prevent this from happening during monitoring, pin the process to specific CPUs (e.g., with SetProcessAffinityMask or start /affinity).

Source code in zeus/device/cpu/emi.py

def get_current_emi_cpu_index() -> int:
    """Return the EMI CPU index of the package the calling thread is running on.

    This is the EMI counterpart of `get_current_rapl_zone_id` in the RAPL module:
    the returned index can be passed as a `cpu_indices` entry to only measure the
    CPU package the current thread is running on.

    The current logical processor (processor group and within-group number) is
    resolved with `GetCurrentProcessorNumberEx` and mapped to a package index
    with `GetLogicalProcessorInformationEx`. Package records are assumed to be
    enumerated in the same order as the EMI `RAPL_Package{N}` channel numbering.

    !!! Note
        The scheduler can migrate threads across packages at any time. To prevent
        this from happening during monitoring, pin the process to specific CPUs
        (e.g., with `SetProcessAffinityMask` or `start /affinity`).
    """
    if not _WINDOWS:
        raise ZeusEMINotSupportedError("EMI is only supported on Windows.")

    proc_number = _ProcessorNumber()
    _kernel32.GetCurrentProcessorNumberEx(ctypes.byref(proc_number))

    # RelationProcessorPackage (= 3) returns one record per physical CPU package.
    relation_processor_package = 3
    buf_size = wintypes.DWORD(0)
    _kernel32.GetLogicalProcessorInformationEx(relation_processor_package, None, ctypes.byref(buf_size))
    if buf_size.value == 0:
        raise RuntimeError(
            f"GetLogicalProcessorInformationEx did not report a buffer size (Windows error {get_last_error()})."
        )
    buf = ctypes.create_string_buffer(buf_size.value)
    if not _kernel32.GetLogicalProcessorInformationEx(relation_processor_package, buf, ctypes.byref(buf_size)):
        raise RuntimeError(f"GetLogicalProcessorInformationEx failed (Windows error {get_last_error()}).")

    package_index = _find_package_of_processor(
        bytes(buf)[: buf_size.value],
        proc_number.Group,
        proc_number.Number,
        ctypes.sizeof(ctypes.c_void_p),
    )
    if package_index is None:
        raise RuntimeError(
            f"Could not map logical processor (group {proc_number.Group}, "
            f"number {proc_number.Number}) to a CPU package."
        )
    return package_index

emi_is_available `cached`

emi_is_available()

Return True if CPU energy can be measured through EMI on this system.

A device only counts as usable if it exposes a RAPL-style package channel (RAPL_Package{N}_PKG), since EMI is also used for other kinds of energy meters (e.g., battery rails) that do not map to CPU packages.

Source code in zeus/device/cpu/emi.py

@lru_cache(maxsize=1)
def emi_is_available() -> bool:
    """Return ``True`` if CPU energy can be measured through EMI on this system.

    A device only counts as usable if it exposes a RAPL-style package channel
    (`RAPL_Package{N}_PKG`), since EMI is also used for other kinds of energy
    meters (e.g., battery rails) that do not map to CPU packages.
    """
    if not _WINDOWS:
        logger.info("EMI is not supported on non-Windows platforms.")
        return False
    try:
        paths = _get_emi_device_paths()
    except Exception as err:
        logger.info("EMI device enumeration failed: %s", err)
        return False
    if not paths:
        logger.info("No EMI energy meter devices found.")
        return False
    for path in paths:
        try:
            emi_file = EMIFile(path)
        except Exception as err:
            logger.info("Cannot open EMI device '%s': %s", path, err)
            continue
        if any(
            _PKG_CHANNEL_RE.match(ch.name) and ch.unit == _EMI_MEASUREMENT_UNIT_PICOWATT_HOURS
            for ch in emi_file.channels
        ):
            logger.info("EMI is available: found a RAPL package channel on '%s'.", path)
            return True
    logger.info(
        "Found %d EMI device(s), but none expose RAPL_Package PKG channels. "
        "CPU energy measurement through EMI is unavailable.",
        len(paths),
    )
    return False

emi

zeus.device.cpu.emi

ZeusEMINotSupportedError

__init__

ZeusEMIInitError

__init__

_EMIChannel

EMIFile

__init__

_read_metadata

_parse_channels staticmethod

read

__del__

EMICPU

__init__

get_total_energy_consumption

supports_get_dram_energy_consumption

EMICPUs

cpus property

__init__

_init_cpus

__del__

_build_emi_guid

_get_emi_device_paths

_ioctl

_find_package_of_processor

get_current_emi_cpu_index

emi_is_available cached

init

init

init

_parse_channels `staticmethod`

del

init

cpus `property`

init

del

emi_is_available `cached`