Skip to content

jetson

zeus.device.soc.jetson

NVIDIA Jetson platform support.

ZeusJetsonInitError

Bases: ZeusSoCInitError

Jetson initialization failures.

Source code in zeus/device/soc/jetson.py 
22
23
24
25
26
27
class ZeusJetsonInitError(ZeusSoCInitError):
    """Jetson initialization failures."""

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

__init__ 

__init__(message)
Source code in zeus/device/soc/jetson.py 
25
26
27
def __init__(self, message: str) -> None:
    """Initialize Zeus Exception."""
    super().__init__(message)

PowerMeasurementStrategy

Bases: ABC

Abstract base class for two different power measurement strategies.

Source code in zeus/device/soc/jetson.py 
35
36
37
38
39
40
41
class PowerMeasurementStrategy(abc.ABC):
    """Abstract base class for two different power measurement strategies."""

    @abc.abstractmethod
    def measure_power(self) -> float:
        """Measure power in mW."""
        pass

measure_power abstractmethod 

measure_power()

Measure power in mW.

Source code in zeus/device/soc/jetson.py 
38
39
40
41
@abc.abstractmethod
def measure_power(self) -> float:
    """Measure power in mW."""
    pass

DirectPower

Bases: PowerMeasurementStrategy

Reads power directly from a sysfs path.

Source code in zeus/device/soc/jetson.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class DirectPower(PowerMeasurementStrategy):
    """Reads power directly from a sysfs path."""

    def __init__(self, power_path: Path) -> None:
        """Initialize DirectPower paths."""
        self.power_path: Path = power_path

    def measure_power(self) -> float:
        """Measure power by reading from sysfs paths.

        Units: mW.
        """
        power: float = float(self.power_path.read_text().strip())
        return power

__init__ 

__init__(power_path)
Source code in zeus/device/soc/jetson.py 
47
48
49
def __init__(self, power_path: Path) -> None:
    """Initialize DirectPower paths."""
    self.power_path: Path = power_path

measure_power 

measure_power()

Measure power by reading from sysfs paths.

Units: mW.

Source code in zeus/device/soc/jetson.py 
51
52
53
54
55
56
57
def measure_power(self) -> float:
    """Measure power by reading from sysfs paths.

    Units: mW.
    """
    power: float = float(self.power_path.read_text().strip())
    return power

VoltageCurrentProduct

Bases: PowerMeasurementStrategy

Computes power as product of voltage and current, read from two sysfs paths.

Source code in zeus/device/soc/jetson.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class VoltageCurrentProduct(PowerMeasurementStrategy):
    """Computes power as product of voltage and current, read from two sysfs paths."""

    def __init__(self, voltage_path: Path, current_path: Path) -> None:
        """Initialize VoltageCurrentProduct paths."""
        self.voltage_path: Path = voltage_path
        self.current_path: Path = current_path

    def measure_power(self) -> float:
        """Measure power by reading from sysfs paths.

        Units: mW.
        """
        voltage: float = float(self.voltage_path.read_text().strip())
        current: float = float(self.current_path.read_text().strip())
        return (voltage * current) / 1000

__init__ 

__init__(voltage_path, current_path)
Source code in zeus/device/soc/jetson.py 
63
64
65
66
def __init__(self, voltage_path: Path, current_path: Path) -> None:
    """Initialize VoltageCurrentProduct paths."""
    self.voltage_path: Path = voltage_path
    self.current_path: Path = current_path

measure_power 

measure_power()

Measure power by reading from sysfs paths.

Units: mW.

Source code in zeus/device/soc/jetson.py 
68
69
70
71
72
73
74
75
def measure_power(self) -> float:
    """Measure power by reading from sysfs paths.

    Units: mW.
    """
    voltage: float = float(self.voltage_path.read_text().strip())
    current: float = float(self.current_path.read_text().strip())
    return (voltage * current) / 1000

JetsonMeasurement dataclass

Bases: SoCMeasurement

Represents energy measurements for Jetson subsystems.

All measurements are in mJ.

Source code in zeus/device/soc/jetson.py 
 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
@dataclass
class JetsonMeasurement(SoCMeasurement):
    """Represents energy measurements for Jetson subsystems.

    All measurements are in mJ.
    """

    cpu_energy_mj: float | None = None
    gpu_energy_mj: float | None = None
    total_energy_mj: float | None = None

    def __sub__(self, other: JetsonMeasurement) -> JetsonMeasurement:
        """Produce a single measurement object containing differences across all fields."""
        if not isinstance(other, type(self)):
            raise TypeError("Subtraction is only supported between Jetson 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
            else:
                setattr(result, f_name, value1 - value2)

        return result

    def zeroAllFields(self) -> None:
        """Set all internal measurement values to zero."""
        for field in fields(self):
            f_name = field.name
            f_value = getattr(self, f_name)
            if isinstance(f_value, float):
                setattr(self, f_name, 0.0)
            else:
                setattr(self, f_name, None)

__sub__ 

__sub__(other)

Produce a single measurement object containing differences across all fields.

Source code in zeus/device/soc/jetson.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def __sub__(self, other: JetsonMeasurement) -> JetsonMeasurement:
    """Produce a single measurement object containing differences across all fields."""
    if not isinstance(other, type(self)):
        raise TypeError("Subtraction is only supported between Jetson 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
        else:
            setattr(result, f_name, value1 - value2)

    return result

zeroAllFields 

zeroAllFields()

Set all internal measurement values to zero.

Source code in zeus/device/soc/jetson.py 
107
108
109
110
111
112
113
114
115
def zeroAllFields(self) -> None:
    """Set all internal measurement values to zero."""
    for field in fields(self):
        f_name = field.name
        f_value = getattr(self, f_name)
        if isinstance(f_value, float):
            setattr(self, f_name, 0.0)
        else:
            setattr(self, f_name, None)

DeviceMap

Bases: TypedDict

Map of device names to their corresponding power measurement strategies.

Source code in zeus/device/soc/jetson.py 
118
119
120
121
122
123
class DeviceMap(TypedDict, total=False):
    """Map of device names to their corresponding power measurement strategies."""

    cpu_power_mw: PowerMeasurementStrategy
    gpu_power_mw: PowerMeasurementStrategy
    total_power_mw: PowerMeasurementStrategy

Jetson

Bases: SoC

An interface for obtaining the energy metrics of a Jetson processor.

Source code in zeus/device/soc/jetson.py 
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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
class Jetson(SoC):
    """An interface for obtaining the energy metrics of a Jetson processor."""

    def __init__(self) -> None:
        """Initialize an instance of a Jetson energy monitor."""
        if not jetson_is_available():
            raise ZeusJetsonInitError(
                "No Jetson processor was detected on the current device."
            )

        super().__init__()

        # Maps each power rail (cpu, gpu, and total) to a power measurement strategy
        self.power_measurement = self._discover_available_metrics()
        self.available_metrics: set[str] | None = None

        # Spawn polling process
        context = mp.get_context("spawn")
        self.command_queue = context.Queue()
        self.result_queue = context.Queue()
        self.process = context.Process(
            target=_polling_process_async_wrapper,
            args=(self.command_queue, self.result_queue, self.power_measurement),
        )
        self.process.start()
        atexit.register(self._stop_process)

    def _discover_available_metrics(self) -> DeviceMap:
        """Return available power measurement metrics per rail from the INA3221 sensor on Jetson devices.

        All official NVIDIA Jetson devices have at least 1 INA3221 power monitor that measures per-rail power usage via 3 channels.

          - https://docs.nvidia.com/jetson/archives/l4t-archived/l4t-3276/index.html#page/Tegra%20Linux%20Driver%20Package%20Development%20Guide/clock_power_setup.html#
          - https://docs.nvidia.com/jetson/archives/r35.6.1/DeveloperGuide/SD/PlatformPowerAndPerformance/JetsonXavierNxSeriesAndJetsonAgxXavierSeries.html#software-based-power-consumption-modeling
          - https://docs.nvidia.com/jetson/archives/r36.4.3/DeveloperGuide/SD/PlatformPowerAndPerformance/JetsonOrinNanoSeriesJetsonOrinNxSeriesAndJetsonAgxOrinSeries.html#
        """
        path = Path("/sys/bus/i2c/drivers/ina3221x")

        metric_paths: dict[str, dict[str, Path]] = {}
        power_measurement: DeviceMap = {}

        def extract_directories(
            path: Path, rail_name: str, rail_index: str, type: str
        ) -> None:
            """Extract file paths for power, voltage, and current measurements based on the rail naming type."""
            rail_name_lower = rail_name.lower()

            if "cpu" in rail_name_lower:
                rail_name_simplified = "cpu_power_mw"
            elif "gpu" in rail_name_lower:
                rail_name_simplified = "gpu_power_mw"
            elif (
                "system" in rail_name_lower
                or "_in" in rail_name_lower
                or "total" in rail_name_lower
            ):
                rail_name_simplified = "total_power_mw"
            else:
                return  # Skip unsupported rail types

            if type == "label":
                power_path = path / f"power{rail_index}_input"
                volt_path = path / f"in{rail_index}_input"
                curr_path = path / f"curr{rail_index}_input"
            else:
                power_path = path / f"in_power{rail_index}_input"
                volt_path = path / f"in_voltage{rail_index}_input"
                curr_path = path / f"in_current{rail_index}_input"

            if check_file(power_path):
                metric_paths[rail_name_simplified] = {"power": Path(power_path)}
            elif check_file(volt_path) and check_file(curr_path):
                metric_paths[rail_name_simplified] = {
                    "volt": Path(volt_path),
                    "curr": Path(curr_path),
                }
            # Else, skip the rail due to insufficient metrics for power

        for device in path.glob("*"):
            for subdevice in device.glob("*"):
                # Get the files containing rail names.
                label_files = subdevice.glob("in*_label")
                rail_files = subdevice.glob("rail_name_*")
                # For each rail name, get its respective power, voltage, current paths.
                for label_file in label_files:
                    rail_name = label_file.read_text().strip()
                    rail_index = label_file.name.split("_")[0].lstrip("in")
                    extract_directories(subdevice, rail_name, rail_index, "label")
                for rail_file in rail_files:
                    rail_name = rail_file.read_text().strip()
                    rail_index = rail_file.name.split("rail_name_", 1)[-1]
                    extract_directories(subdevice, rail_name, rail_index, "rail_name")

        # Instantiate PowerMeasurementStrategy objects based on available metrics
        for rail, metrics in metric_paths.items():
            if "power" in metrics:
                power_measurement[rail] = DirectPower(metrics["power"])
            elif "volt" in metrics and "curr" in metrics:
                power_measurement[rail] = VoltageCurrentProduct(
                    metrics["volt"], metrics["curr"]
                )
            # Else, skip the rail due to insufficient metrics for power
        return power_measurement

    def getAvailableMetrics(self) -> set[str]:
        """Return a set of all observable metrics on the Jetson device."""
        if self.available_metrics is None:
            result: JetsonMeasurement = self.getTotalEnergyConsumption()
            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 _stop_process(self) -> None:
        """Kill the polling process."""
        self.command_queue.put_nowait(Command.STOP)
        self.process.join(timeout=1.0)
        self.process.kill()

    def getTotalEnergyConsumption(self, timeout: float = 15.0) -> JetsonMeasurement:
        """Returns the total energy consumption of the Jetson device. This measurement is cumulative.

        Units: mJ.
        """
        self.command_queue.put(Command.READ)
        return self.result_queue.get(timeout=timeout)

__init__ 

__init__()
Source code in zeus/device/soc/jetson.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def __init__(self) -> None:
    """Initialize an instance of a Jetson energy monitor."""
    if not jetson_is_available():
        raise ZeusJetsonInitError(
            "No Jetson processor was detected on the current device."
        )

    super().__init__()

    # Maps each power rail (cpu, gpu, and total) to a power measurement strategy
    self.power_measurement = self._discover_available_metrics()
    self.available_metrics: set[str] | None = None

    # Spawn polling process
    context = mp.get_context("spawn")
    self.command_queue = context.Queue()
    self.result_queue = context.Queue()
    self.process = context.Process(
        target=_polling_process_async_wrapper,
        args=(self.command_queue, self.result_queue, self.power_measurement),
    )
    self.process.start()
    atexit.register(self._stop_process)

_discover_available_metrics 

_discover_available_metrics()

Return available power measurement metrics per rail from the INA3221 sensor on Jetson devices.

All official NVIDIA Jetson devices have at least 1 INA3221 power monitor that measures per-rail power usage via 3 channels.

  • https://docs.nvidia.com/jetson/archives/l4t-archived/l4t-3276/index.html#page/Tegra%20Linux%20Driver%20Package%20Development%20Guide/clock_power_setup.html#
  • https://docs.nvidia.com/jetson/archives/r35.6.1/DeveloperGuide/SD/PlatformPowerAndPerformance/JetsonXavierNxSeriesAndJetsonAgxXavierSeries.html#software-based-power-consumption-modeling
  • https://docs.nvidia.com/jetson/archives/r36.4.3/DeveloperGuide/SD/PlatformPowerAndPerformance/JetsonOrinNanoSeriesJetsonOrinNxSeriesAndJetsonAgxOrinSeries.html#
Source code in zeus/device/soc/jetson.py 
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
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def _discover_available_metrics(self) -> DeviceMap:
    """Return available power measurement metrics per rail from the INA3221 sensor on Jetson devices.

    All official NVIDIA Jetson devices have at least 1 INA3221 power monitor that measures per-rail power usage via 3 channels.

      - https://docs.nvidia.com/jetson/archives/l4t-archived/l4t-3276/index.html#page/Tegra%20Linux%20Driver%20Package%20Development%20Guide/clock_power_setup.html#
      - https://docs.nvidia.com/jetson/archives/r35.6.1/DeveloperGuide/SD/PlatformPowerAndPerformance/JetsonXavierNxSeriesAndJetsonAgxXavierSeries.html#software-based-power-consumption-modeling
      - https://docs.nvidia.com/jetson/archives/r36.4.3/DeveloperGuide/SD/PlatformPowerAndPerformance/JetsonOrinNanoSeriesJetsonOrinNxSeriesAndJetsonAgxOrinSeries.html#
    """
    path = Path("/sys/bus/i2c/drivers/ina3221x")

    metric_paths: dict[str, dict[str, Path]] = {}
    power_measurement: DeviceMap = {}

    def extract_directories(
        path: Path, rail_name: str, rail_index: str, type: str
    ) -> None:
        """Extract file paths for power, voltage, and current measurements based on the rail naming type."""
        rail_name_lower = rail_name.lower()

        if "cpu" in rail_name_lower:
            rail_name_simplified = "cpu_power_mw"
        elif "gpu" in rail_name_lower:
            rail_name_simplified = "gpu_power_mw"
        elif (
            "system" in rail_name_lower
            or "_in" in rail_name_lower
            or "total" in rail_name_lower
        ):
            rail_name_simplified = "total_power_mw"
        else:
            return  # Skip unsupported rail types

        if type == "label":
            power_path = path / f"power{rail_index}_input"
            volt_path = path / f"in{rail_index}_input"
            curr_path = path / f"curr{rail_index}_input"
        else:
            power_path = path / f"in_power{rail_index}_input"
            volt_path = path / f"in_voltage{rail_index}_input"
            curr_path = path / f"in_current{rail_index}_input"

        if check_file(power_path):
            metric_paths[rail_name_simplified] = {"power": Path(power_path)}
        elif check_file(volt_path) and check_file(curr_path):
            metric_paths[rail_name_simplified] = {
                "volt": Path(volt_path),
                "curr": Path(curr_path),
            }
        # Else, skip the rail due to insufficient metrics for power

    for device in path.glob("*"):
        for subdevice in device.glob("*"):
            # Get the files containing rail names.
            label_files = subdevice.glob("in*_label")
            rail_files = subdevice.glob("rail_name_*")
            # For each rail name, get its respective power, voltage, current paths.
            for label_file in label_files:
                rail_name = label_file.read_text().strip()
                rail_index = label_file.name.split("_")[0].lstrip("in")
                extract_directories(subdevice, rail_name, rail_index, "label")
            for rail_file in rail_files:
                rail_name = rail_file.read_text().strip()
                rail_index = rail_file.name.split("rail_name_", 1)[-1]
                extract_directories(subdevice, rail_name, rail_index, "rail_name")

    # Instantiate PowerMeasurementStrategy objects based on available metrics
    for rail, metrics in metric_paths.items():
        if "power" in metrics:
            power_measurement[rail] = DirectPower(metrics["power"])
        elif "volt" in metrics and "curr" in metrics:
            power_measurement[rail] = VoltageCurrentProduct(
                metrics["volt"], metrics["curr"]
            )
        # Else, skip the rail due to insufficient metrics for power
    return power_measurement

getAvailableMetrics 

getAvailableMetrics()

Return a set of all observable metrics on the Jetson device.

Source code in zeus/device/soc/jetson.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
def getAvailableMetrics(self) -> set[str]:
    """Return a set of all observable metrics on the Jetson device."""
    if self.available_metrics is None:
        result: JetsonMeasurement = self.getTotalEnergyConsumption()
        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

_stop_process 

_stop_process()

Kill the polling process.

Source code in zeus/device/soc/jetson.py 
244
245
246
247
248
def _stop_process(self) -> None:
    """Kill the polling process."""
    self.command_queue.put_nowait(Command.STOP)
    self.process.join(timeout=1.0)
    self.process.kill()

getTotalEnergyConsumption 

getTotalEnergyConsumption(timeout=15.0)

Returns the total energy consumption of the Jetson device. This measurement is cumulative.

Units: mJ.

Source code in zeus/device/soc/jetson.py 
250
251
252
253
254
255
256
def getTotalEnergyConsumption(self, timeout: float = 15.0) -> JetsonMeasurement:
    """Returns the total energy consumption of the Jetson device. This measurement is cumulative.

    Units: mJ.
    """
    self.command_queue.put(Command.READ)
    return self.result_queue.get(timeout=timeout)

Command

Bases: Enum

Provide commands for the polling process.

Source code in zeus/device/soc/jetson.py 
259
260
261
262
263
class Command(enum.Enum):
    """Provide commands for the polling process."""

    READ = "read"
    STOP = "stop"

check_file 

check_file(path)

Check if the given path exists and is a file.

Source code in zeus/device/soc/jetson.py 
30
31
32
def check_file(path: Path) -> bool:
    """Check if the given path exists and is a file."""
    return path.exists() and path.is_file()

_polling_process_async_wrapper 

_polling_process_async_wrapper(
    command_queue, result_queue, power_measurement
)

Function wrapper for the asynchronous energy polling process.

Source code in zeus/device/soc/jetson.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
def _polling_process_async_wrapper(
    command_queue: mp.Queue[Command],
    result_queue: mp.Queue[JetsonMeasurement],
    power_measurement: DeviceMap,
) -> None:
    """Function wrapper for the asynchronous energy polling process."""
    asyncio.run(
        _polling_process_async(
            command_queue,
            result_queue,
            power_measurement,
        )
    )

_polling_process_async async 

_polling_process_async(
    command_queue, result_queue, power_measurement
)

Continuously polls for accumulated energy measurements for CPU, GPU, and total power, listening for commands to stop or return the measurement.

Source code in zeus/device/soc/jetson.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
async def _polling_process_async(
    command_queue: mp.Queue[Command],
    result_queue: mp.Queue[JetsonMeasurement],
    power_measurement: DeviceMap,
) -> None:
    """Continuously polls for accumulated energy measurements for CPU, GPU, and total power, listening for commands to stop or return the measurement."""
    cumulative_measurement = JetsonMeasurement(
        cpu_energy_mj=0.0 if "cpu_power_mw" in power_measurement else None,
        gpu_energy_mj=0.0 if "gpu_power_mw" in power_measurement else None,
        total_energy_mj=0.0 if "total_power_mw" in power_measurement else None,
    )

    prev_ts = time.monotonic()

    while True:
        current_ts: float = time.monotonic()
        dt: float = current_ts - prev_ts

        if "cpu_power_mw" in power_measurement:
            cpu_power_mw = power_measurement["cpu_power_mw"].measure_power()
            cpu_energy_mj = cpu_power_mw * dt
            cumulative_measurement.cpu_energy_mj = (
                cumulative_measurement.cpu_energy_mj or 0.0
            ) + cpu_energy_mj
        if "gpu_power_mw" in power_measurement:
            gpu_power_mw = power_measurement["gpu_power_mw"].measure_power()
            gpu_energy_mj = gpu_power_mw * dt
            cumulative_measurement.gpu_energy_mj = (
                cumulative_measurement.gpu_energy_mj or 0.0
            ) + gpu_energy_mj
        if "total_power_mw" in power_measurement:
            total_power_mw = power_measurement["total_power_mw"].measure_power()
            total_energy_mj = total_power_mw * dt
            cumulative_measurement.total_energy_mj = (
                cumulative_measurement.total_energy_mj or 0.0
            ) + total_energy_mj

        prev_ts = current_ts

        try:
            command = await asyncio.to_thread(
                command_queue.get,
                timeout=0.1,
            )
        except Empty:
            # Update energy and do nothing
            continue

        if command == Command.STOP:
            break
        if command == Command.READ:
            # Update and return energy measurement
            result_queue.put(cumulative_measurement)

jetson_is_available 

jetson_is_available()

Return if the current processor is a Jetson device.

Source code in zeus/device/soc/jetson.py 
336
337
338
339
340
341
342
343
def jetson_is_available() -> bool:
    """Return if the current processor is a Jetson device."""
    if sys.platform != "linux" or platform.processor() != "aarch64":
        return False

    return os.path.exists("/usr/lib/aarch64-linux-gnu/tegra") or os.path.exists(
        "/etc/nv_tegra_release"
    )