# SPDX-FileCopyrightText: Copyright (c) 2022-2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: LicenseRef-NvidiaProprietary # # NVIDIA CORPORATION, its affiliates and licensors retain all intellectual # property and proprietary rights in and to this material, related # documentation and any modifications thereto. Any use, reproduction, # disclosure or distribution of this material and related documentation # without an express license agreement from NVIDIA CORPORATION or # its affiliates is strictly prohibited. """This package is responsible for updating power data periodically. Data center plays the role of subject in the observer pattern, it will maintain a list of its dependents (observers) and notify them when the power data is updated. MainThread is designed for rendering GUI, so we want to reduce the burden on it. Hence, we dispatch the data update tasks to another daemon thread (called DataThread). Data center plays the role of a provider, all update tasks will be periodically pushed to the work queue by it. In the other hand, the DataThread will play the role of a consumer, take out tasks in the workqueue and execute them. """ import subprocess from collections import OrderedDict from datetime import datetime from pylibjetsonpower import * # Import all public power APIs from ..common import ( PAttrs, PNames, err_symbol, kHz2mHz, millC2C, tsensor_off_ret, ) from ..subjectobserver import Subject from ..workqueue import work_queue class PowerData(OrderedDict): def __init__(self, **kwargs): super().__init__(**kwargs) def __missing__(self, key): # Retrun err_symbol on KeyError return err_symbol class DataCenter(Subject): cpu_nums = cpu_get_nums() igpu_nums = 1 # TODO: consider dGPU engine_nums = len(PNames.ENGINE_NAMES) sensor_nums = len(PNames.THERMAL_SENSOR_NAMES) rail_nums = len(PNames.RAIL_NAMES) fan_nums = len(PNames.FAN_NAMES) disk_df = subprocess.run(["df", "/", "--output=source"], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) disk_tail = subprocess.run(["tail", "-n", "1"], check=True, input=disk_df.stdout, stdout=subprocess.PIPE, stderr=subprocess.PIPE) disk_sed = subprocess.run(["sed", "s/\\/dev\\///"], check=True, input=disk_tail.stdout, stdout=subprocess.PIPE, stderr=subprocess.PIPE) disk_name = str(disk_sed.stdout, "utf-8").strip() def __init__(self): super().__init__() self.data = PowerData() self.observers = [] # observers to be notified when power data updates def attach(self, observer): self.observers.append(observer) def detach(self, observer): self.observers.remove(observer) def notify(self): for observer in self.observers: observer.update_data(self.data) def get_data(self): return self.data def set_data(self, key, val): self.data[key] = val def del_data(self, key): if key in self.data: self.data.pop(key) def update_data(self): # Since power data is an ordered dict, so it keeps the insertion order # of keys. In current design, the order of the heading in the log file # depeneds on the order that update_xxx() get called here. work_queue.add_work(self.update_time) work_queue.add_work(self.update_nvpmodel) work_queue.add_work(self.update_cpu) work_queue.add_work(self.update_gpu) work_queue.add_work(self.update_emc) work_queue.add_work(self.update_engine) work_queue.add_work(self.update_thermal) work_queue.add_work(self.update_power_monitor) work_queue.add_work(self.update_fan) work_queue.add_work(self.update_disk) work_queue.add_work(self.notify) def handle_return(self, val, deco_f=lambda x: x, check_f=lambda x: x >= 0): return deco_f(val) if check_f(val) else err_symbol def update_time(self): self.set_data(("System", "Time"), datetime.now().strftime("%H:%M:%S")) def update_nvpmodel(self): self.set_data((PNames.NVPMODEL, PAttrs.POWER_MODE), self.handle_return(nvpmodel_get_power_mode(), check_f=lambda x: bool(x))) def update_cpu(self): for i in range(DataCenter.cpu_nums): name = f"{PNames.CPU}{i}" if 1 == self.handle_return(cpu_get_state(i)): self.set_data((name, PAttrs.STATE), True) self.set_data((name, PAttrs.CURR_FREQ), self.handle_return(cpu_get_cur_freq(i), kHz2mHz)) self.set_data((name, PAttrs.CURR_LOAD), self.handle_return(cpu_get_load(i))) else: self.set_data((name, PAttrs.STATE), False) self.set_data((name, PAttrs.CURR_FREQ), err_symbol) self.set_data((name, PAttrs.CURR_LOAD), err_symbol) def update_gpu(self): for i in range(DataCenter.igpu_nums): name = f"{PNames.IGPU}{i}" self.set_data((name, PAttrs.CURR_FREQ), self.handle_return(igpu_get_cur_freq(i), kHz2mHz)) self.set_data((name, PAttrs.CURR_LOAD), self.handle_return(igpu_get_load(i))) def update_emc(self): total_ram = self.handle_return(emc_get_mem_size()) free_ram = self.handle_return(emc_get_free_mem_size()) buffers_ram = self.handle_return(emc_get_buffers_size()) cached_ram = self.handle_return(emc_get_cached_size()) total_swap = self.handle_return(emc_get_swap_size()) free_swap = self.handle_return(emc_get_free_swap_size()) mem_used = err_symbol swap_used = err_symbol mem_load = err_symbol swap_load = err_symbol if err_symbol not in (total_ram, free_ram, buffers_ram, cached_ram): if total_ram > 0: mem_used = total_ram - free_ram - buffers_ram - cached_ram mem_load = round(100 * mem_used / total_ram) if err_symbol not in (total_swap, free_swap): if total_swap > 0: swap_used = total_swap - free_swap swap_load = round(100 * swap_used / total_swap) self.set_data((PNames.EMC, PAttrs.CURR_FREQ), self.handle_return(emc_get_cur_freq(), kHz2mHz)) self.set_data((PNames.EMC, PAttrs.CURR_LOAD), self.handle_return(emc_get_load())) self.set_data((PNames.MEM, PAttrs.USED_SIZE), mem_used) self.set_data((PNames.MEM, PAttrs.TOTAL_SIZE), total_ram) self.set_data((PNames.MEM, PAttrs.CURR_LOAD), mem_load) self.set_data((PNames.SWAP, PAttrs.USED_SIZE), swap_used) self.set_data((PNames.SWAP, PAttrs.TOTAL_SIZE), total_swap) self.set_data((PNames.SWAP, PAttrs.CURR_LOAD), swap_load) def update_engine(self): for i in range(DataCenter.engine_nums): name = PNames.ENGINE_NAMES[i] if 1 == self.handle_return(engine_get_state(name)): self.set_data((name, PAttrs.STATE), True) self.set_data((name, PAttrs.CURR_FREQ), self.handle_return(engine_get_cur_freq(name), kHz2mHz)) else: self.set_data((name, PAttrs.STATE), False) self.set_data((name, PAttrs.CURR_FREQ), err_symbol) def update_thermal(self): for i in range(DataCenter.sensor_nums): name = PNames.THERMAL_SENSOR_NAMES[i] curr_temp = thermal_get_sensor_temp(name) if tsensor_off_ret == curr_temp: # Propagate the error code if the sensor's power domain is off self.set_data((name, PAttrs.CURR_TEMP), tsensor_off_ret) else: self.set_data((name, PAttrs.CURR_TEMP), self.handle_return(curr_temp, millC2C, lambda x: thermal_is_temp_valid(x))) self.set_data( (name, PAttrs.SW_THROTTLING_TEMP), self.handle_return( thermal_get_sensor_sw_throttling_temp(name), millC2C, lambda x: thermal_is_temp_valid(x))) self.set_data( (name, PAttrs.SW_SHUTDOWN_TEMP), self.handle_return( thermal_get_sensor_sw_shutdown_temp(name), millC2C, lambda x: thermal_is_temp_valid(x))) def update_power_monitor(self): for i in range(DataCenter.rail_nums): name = PNames.RAIL_NAMES[i] self.set_data((name, PAttrs.WARN_CURRENT), self.handle_return(rail_get_warn_current(name))) self.set_data((name, PAttrs.CRIT_CURRENT), self.handle_return(rail_get_crit_current(name))) self.set_data((name, PAttrs.IN_VOLTAGE), self.handle_return(rail_get_voltage(name))) self.set_data((name, PAttrs.IN_CURRENT), self.handle_return(rail_get_current(name))) self.set_data((name, PAttrs.IN_POWER), self.handle_return(rail_get_power(name))) self.set_data((name, PAttrs.AVG_POWER), self.handle_return(rail_get_avg_power(name))) def update_fan(self): for i in range(DataCenter.fan_nums): fan_name = PNames.FAN_NAMES[i] self.set_data((fan_name, PAttrs.FAN_PROFILE), self.handle_return(fan_get_profile(i + 1), check_f=lambda x: bool(x))) self.set_data((fan_name, PAttrs.FAN_GOVERNOR), self.handle_return(fan_get_governor(i + 1), check_f=lambda x: bool(x))) self.set_data((fan_name, PAttrs.FAN_CONTROL), self.handle_return(fan_get_control(i + 1), check_f=lambda x: bool(x))) self.set_data((fan_name, PAttrs.CURR_PWM), self.handle_return(fan_get_pwm(i + 1))) self.set_data((fan_name, PAttrs.RPM_MEASURED), self.handle_return(fan_get_speed(i + 1))) def update_disk(self): self.set_data((PNames.DISK, PAttrs.USED_SIZE), self.handle_return(disk_get_used_size(DataCenter.disk_name))) self.set_data((PNames.DISK, PAttrs.TOTAL_SIZE), self.handle_return(disk_get_total_size(DataCenter.disk_name))) # Instantiate singleton data_center = DataCenter()