Skip to content

apple

zeus.device.soc.apple

Apple Silicon SoC's.

MockZeusAppleSilicon

Mock class for zeus-apple-silicon library.

Source code in zeus/device/soc/apple.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
class MockZeusAppleSilicon:
    """Mock class for zeus-apple-silicon library."""

    def __getattr__(self, name):
        """Raise an error if any method is called.

        Since this class is only used when `zeus-apple-silicon` is not
        available, something has gone wrong if any method is called.
        """
        raise RuntimeError(
            f"zeus-apple-silicon is not available and zeus-apple-silicon.{name} "
            "shouldn't have been called. This is a bug."
        )

__getattr__ 

__getattr__(name)

Raise an error if any method is called.

Since this class is only used when zeus-apple-silicon is not available, something has gone wrong if any method is called.

Source code in zeus/device/soc/apple.py 
22
23
24
25
26
27
28
29
30
31
def __getattr__(self, name):
    """Raise an error if any method is called.

    Since this class is only used when `zeus-apple-silicon` is not
    available, something has gone wrong if any method is called.
    """
    raise RuntimeError(
        f"zeus-apple-silicon is not available and zeus-apple-silicon.{name} "
        "shouldn't have been called. This is a bug."
    )

ZeusAppleInitError

Bases: ZeusSoCInitError

Import error for Apple SoC initialization failures.

Source code in zeus/device/soc/apple.py 
47
48
49
50
51
52
class ZeusAppleInitError(ZeusSoCInitError):
    """Import error for Apple SoC initialization failures."""

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

__init__ 

__init__(message)
Source code in zeus/device/soc/apple.py 
50
51
52
def __init__(self, message: str) -> None:
    """Initialize Zeus Exception."""
    super().__init__(message)

AppleSiliconMeasurement dataclass

Bases: SoCMeasurement

Represents energy consumption of various subsystems on an Apple processor.

All measurements are in mJ.

Source code in zeus/device/soc/apple.py 
 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
@dataclass
class AppleSiliconMeasurement(SoCMeasurement):
    """Represents energy consumption of various subsystems on an Apple processor.

    All measurements are in mJ.
    """

    # CPU related metrics
    cpu_total_mj: int | None = None
    efficiency_cores_mj: list[int] | None = None
    performance_cores_mj: list[int] | None = None
    efficiency_core_manager_mj: int | None = None
    performance_core_manager_mj: int | None = None

    # DRAM
    dram_mj: int | None = None

    # GPU related metrics
    gpu_mj: int | None = None
    gpu_sram_mj: int | None = None

    # ANE (Apple Neural Engine)
    ane_mj: int | None = None

    def __sub__(self, other: AppleSiliconMeasurement) -> AppleSiliconMeasurement:
        """Produce a single measurement object containing differences across all fields."""
        if not isinstance(other, type(self)):
            raise TypeError("Subtraction is only supported between AppleSiliconMeasurement instances.")

        result = self.__class__()

        for field in fields(self):
            f_name = field.name
            value1 = getattr(self, f_name)
            value2 = getattr(other, f_name)
            if value1 is None and value2 is None:
                continue

            if type(value1) is not type(value2):
                raise ValueError(f"Inconsistent field between two AppleSiliconMeasurement objects: {f_name}")

            if isinstance(value1, int):
                setattr(result, f_name, value1 - value2)
            elif isinstance(value1, list):
                if len(value1) != len(value2):
                    raise ValueError(f"Inconsistent field between two AppleSiliconMeasurement objects: {f_name}")
                setattr(result, f_name, [x - y for x, y in zip(value1, value2)])

        return result

    def zero_all_fields(self) -> None:
        """Set the value of all fields in the measurement object to zero."""
        for field in fields(self):
            f_name = field.name
            f_value = getattr(self, f_name)
            if isinstance(f_value, int):
                setattr(self, f_name, 0)
            elif isinstance(f_value, list):
                setattr(self, f_name, [])
            else:
                setattr(self, f_name, None)

    @classmethod
    def from_metrics(
        cls,
        metrics: zeus_apple_silicon.AppleEnergyMetrics,  # type: ignore
    ) -> AppleSiliconMeasurement:
        """Return an AppleSiliconMeasurement object based on an AppleEnergyMetrics object."""
        return cls(
            cpu_total_mj=metrics.cpu_total_mj,
            efficiency_cores_mj=metrics.efficiency_cores_mj,
            performance_cores_mj=metrics.performance_cores_mj,
            efficiency_core_manager_mj=metrics.efficiency_core_manager_mj,
            performance_core_manager_mj=metrics.performance_core_manager_mj,
            dram_mj=metrics.dram_mj,
            gpu_mj=metrics.gpu_mj,
            gpu_sram_mj=metrics.gpu_sram_mj,
            ane_mj=metrics.ane_mj,
        )

__sub__ 

__sub__(other)

Produce a single measurement object containing differences across all fields.

Source code in zeus/device/soc/apple.py 
 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
def __sub__(self, other: AppleSiliconMeasurement) -> AppleSiliconMeasurement:
    """Produce a single measurement object containing differences across all fields."""
    if not isinstance(other, type(self)):
        raise TypeError("Subtraction is only supported between AppleSiliconMeasurement instances.")

    result = self.__class__()

    for field in fields(self):
        f_name = field.name
        value1 = getattr(self, f_name)
        value2 = getattr(other, f_name)
        if value1 is None and value2 is None:
            continue

        if type(value1) is not type(value2):
            raise ValueError(f"Inconsistent field between two AppleSiliconMeasurement objects: {f_name}")

        if isinstance(value1, int):
            setattr(result, f_name, value1 - value2)
        elif isinstance(value1, list):
            if len(value1) != len(value2):
                raise ValueError(f"Inconsistent field between two AppleSiliconMeasurement objects: {f_name}")
            setattr(result, f_name, [x - y for x, y in zip(value1, value2)])

    return result

zero_all_fields 

zero_all_fields()

Set the value of all fields in the measurement object to zero.

Source code in zeus/device/soc/apple.py 
105
106
107
108
109
110
111
112
113
114
115
def zero_all_fields(self) -> None:
    """Set the value of all fields in the measurement object to zero."""
    for field in fields(self):
        f_name = field.name
        f_value = getattr(self, f_name)
        if isinstance(f_value, int):
            setattr(self, f_name, 0)
        elif isinstance(f_value, list):
            setattr(self, f_name, [])
        else:
            setattr(self, f_name, None)

from_metrics classmethod 

from_metrics(metrics)

Return an AppleSiliconMeasurement object based on an AppleEnergyMetrics object.

Source code in zeus/device/soc/apple.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@classmethod
def from_metrics(
    cls,
    metrics: zeus_apple_silicon.AppleEnergyMetrics,  # type: ignore
) -> AppleSiliconMeasurement:
    """Return an AppleSiliconMeasurement object based on an AppleEnergyMetrics object."""
    return cls(
        cpu_total_mj=metrics.cpu_total_mj,
        efficiency_cores_mj=metrics.efficiency_cores_mj,
        performance_cores_mj=metrics.performance_cores_mj,
        efficiency_core_manager_mj=metrics.efficiency_core_manager_mj,
        performance_core_manager_mj=metrics.performance_core_manager_mj,
        dram_mj=metrics.dram_mj,
        gpu_mj=metrics.gpu_mj,
        gpu_sram_mj=metrics.gpu_sram_mj,
        ane_mj=metrics.ane_mj,
    )

AppleSilicon

Bases: SoC

An interface for obtaining energy metrics of an Apple processor.

Source code in zeus/device/soc/apple.py 
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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
class AppleSilicon(SoC):
    """An interface for obtaining energy metrics of an Apple processor."""

    def __init__(self) -> None:
        """Initialize an instance of an Apple Silicon energy monitor."""
        self._monitor: zeus_apple_silicon.AppleEnergyMonitor  # type: ignore
        self.available_metrics: set[str] | None = None

        try:
            self._monitor = zeus_apple_silicon.AppleEnergyMonitor()

        # This except block exists for failures the AppleEnergyMonitor
        # object may encounter during its own construction.
        except RuntimeError as e:
            raise ZeusAppleInitError(f"Failed to initialize `AppleEnergyMonitor`: {e}") from None

    def get_available_metrics(self) -> set[str]:
        """Return a set of all observable metrics on the current processor."""
        if self.available_metrics is None:
            result: SoCMeasurement = self.get_total_energy_consumption()
            available_metrics = set()

            metrics_dict = asdict(result)
            for f_name, f_value in metrics_dict.items():
                if f_value is not None:
                    available_metrics.add(f_name)

            self.available_metrics = available_metrics
        return self.available_metrics

    def get_total_energy_consumption(self) -> AppleSiliconMeasurement:
        """Returns the total energy consumption of the SoC.

        The measurement should be cumulative; different calls to this function throughout
        the lifetime of a single `SoC` manager object should count from a fixed arbitrary
        point in time.

        Units: mJ.
        """
        result = self._monitor.get_cumulative_energy()
        return AppleSiliconMeasurement.from_metrics(result)

    def begin_window(self, key) -> None:
        """Begin a measurement interval labeled with `key`."""
        self._monitor.begin_window(key)

    def end_window(self, key) -> AppleSiliconMeasurement:
        """End a measurement window and return the energy consumption. Units: mJ."""
        result = self._monitor.end_window(key)
        return AppleSiliconMeasurement.from_metrics(result)

__init__ 

__init__()
Source code in zeus/device/soc/apple.py 
139
140
141
142
143
144
145
146
147
148
149
150
def __init__(self) -> None:
    """Initialize an instance of an Apple Silicon energy monitor."""
    self._monitor: zeus_apple_silicon.AppleEnergyMonitor  # type: ignore
    self.available_metrics: set[str] | None = None

    try:
        self._monitor = zeus_apple_silicon.AppleEnergyMonitor()

    # This except block exists for failures the AppleEnergyMonitor
    # object may encounter during its own construction.
    except RuntimeError as e:
        raise ZeusAppleInitError(f"Failed to initialize `AppleEnergyMonitor`: {e}") from None

get_available_metrics 

get_available_metrics()

Return a set of all observable metrics on the current processor.

Source code in zeus/device/soc/apple.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
def get_available_metrics(self) -> set[str]:
    """Return a set of all observable metrics on the current processor."""
    if self.available_metrics is None:
        result: SoCMeasurement = self.get_total_energy_consumption()
        available_metrics = set()

        metrics_dict = asdict(result)
        for f_name, f_value in metrics_dict.items():
            if f_value is not None:
                available_metrics.add(f_name)

        self.available_metrics = available_metrics
    return self.available_metrics

get_total_energy_consumption 

get_total_energy_consumption()

Returns the total energy consumption of the SoC.

The measurement should be cumulative; different calls to this function throughout the lifetime of a single SoC manager object should count from a fixed arbitrary point in time.

Units: mJ.

Source code in zeus/device/soc/apple.py 
166
167
168
169
170
171
172
173
174
175
176
def get_total_energy_consumption(self) -> AppleSiliconMeasurement:
    """Returns the total energy consumption of the SoC.

    The measurement should be cumulative; different calls to this function throughout
    the lifetime of a single `SoC` manager object should count from a fixed arbitrary
    point in time.

    Units: mJ.
    """
    result = self._monitor.get_cumulative_energy()
    return AppleSiliconMeasurement.from_metrics(result)

begin_window 

begin_window(key)

Begin a measurement interval labeled with key.

Source code in zeus/device/soc/apple.py 
178
179
180
def begin_window(self, key) -> None:
    """Begin a measurement interval labeled with `key`."""
    self._monitor.begin_window(key)

end_window 

end_window(key)

End a measurement window and return the energy consumption. Units: mJ.

Source code in zeus/device/soc/apple.py 
182
183
184
185
def end_window(self, key) -> AppleSiliconMeasurement:
    """End a measurement window and return the energy consumption. Units: mJ."""
    result = self._monitor.end_window(key)
    return AppleSiliconMeasurement.from_metrics(result)

apple_silicon_is_available cached 

apple_silicon_is_available()

Check if Apple silicon is available.

Source code in zeus/device/soc/apple.py 
37
38
39
40
41
42
43
44
@lru_cache(maxsize=1)
def apple_silicon_is_available() -> bool:
    """Check if Apple silicon is available."""
    if not zeus_apple_available:
        return False
    if sys.platform != "darwin" or platform.processor() != "arm":
        return False
    return True