dell-fans-controller-docker/controller/ipmi.py

import subprocess
import time
import re
from controller.logger import logger

class IpmiTool:
    def __init__(self, host: str, username: str, password: str):
        if not host or not username or not password:
            raise ValueError("host, username and password must be provided")
        self.host = host
        self.username = username
        self.password = password

    def run_cmd(self, cmd: str) -> str:
        basecmd = f'ipmitool -H {self.host} -I lanplus -U {self.username} -P {self.password}'
        command = f'{basecmd} {cmd}'
        retry_count = 3  # 设置重试次数
        for attempt in range(retry_count):
            try:
                # print(f"Executing command: {command}")  # 添加调试信息
                result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)

                if result.returncode != 0:
                    raise RuntimeError(
                        f'IPMI 命令执行失败: {cmd}\n错误详情: {result.stderr}'  # 更清晰的错误提示
                    )
                    # 添加网络和认证排查提示
                    print("请检查以下内容：")
                    print("1. 确保 BMC 地址可访问（ping 测试或网络配置）。")
                    print("2. 验证用户名、密码是否正确。")
                    print("3. 检查目标设备的 IPMI 功能是否启用。")

                return result.stdout
            except subprocess.TimeoutExpired:
                if attempt < retry_count - 1:
                    logger.warning(f'命令超时，正在重试... (尝试次数 {attempt + 1}/{retry_count})')
                    time.sleep(5)  # 每次重试前等待 5 秒
                else:
                    raise RuntimeError('IPMI 命令超时。请检查网络连接或服务器状态。')  # 更明确的错误提示

    def mc_info(self) -> str:
        """
        execute ipmitool command mc info
        :return:
        """
        return self.run_cmd(cmd='mc info')

    def sensor(self) -> str:
        """
        execute ipmitool command sdr to get sensor data
        :return:
        """
        return self.run_cmd(cmd='sdr')

    def temperature(self) -> list:
        """
        get current temperature
        :return:
        """
        data = self.sensor()
        temperatures = []
        import re

        for line in data.splitlines():
            if 'Temp' in line and 'degrees C' in line:
                # 提取温度值，例如从 " 25 degrees C" 中提取 25
                temp_part = line.split('|')[1]  # 获取中间列的内容
                # 使用正则表达式提取数字
                match = re.search(r'(\d+(\.\d+)?)\s+degrees C', temp_part)
                if match:
                    temp_value = float(match.group(1))
                    temperatures.append(temp_value)

        return temperatures

    def fan_speeds(self) -> list:
        """
        get current fan speeds
        :return: list of fan speeds in percentage
        """
        data = self.sensor()
        fan_speeds = []

        for line in data.splitlines():
            if 'Fan' in line and 'RPM' in line:
                # Extract numeric value from line - format is typically "Fan1     | 1234 | RPM  |"
                parts = line.split('|')
                if len(parts) >= 2:
                    try:
                        # Extract the value and convert RPM to percentage if possible
                        # For Dell servers, we may need to get duty cycle instead
                        value_str = parts[1].strip()
                        if value_str.isdigit():
                            rpm = int(value_str)
                            # Placeholder: we might need to use raw commands to get duty cycle
                            # For now, return the raw value
                            fan_speeds.append(rpm)
                    except ValueError:
                        continue
        return fan_speeds

    def get_fan_duty_cycle(self) -> int:
        """
        get current fan duty cycle/percentage
        :return: current fan duty cycle in percentage
        """
        try:
            # Raw command to get current fan duty cycle
            result = self.run_cmd('raw 0x30 0x31 0x01')
            # Parse the hex result to get duty cycle
            result_parts = result.strip().split()
            if result_parts and len(result_parts) >= 1:
                # The command should return a hex value representing the duty cycle
                duty_cycle_hex = result_parts[-1]
                duty_cycle = int(duty_cycle_hex, 16)
                # Ensure the value is in valid range (0-100)
                if 0 <= duty_cycle <= 100 and duty_cycle != 0:
                    # If we get a reasonable value (not 0), return it
                    return duty_cycle
                elif duty_cycle == 0:
                    # Value of 0 might indicate auto mode or that raw command doesn't return duty cycle on this system
                    logger.info('原始命令返回0，尝试从RPM估算风扇百分比')
        except Exception as e:
            logger.warning(f'获取风扇占空比的原始命令失败: {e}')

        # If raw command fails or returns 0, get fan speeds from sensor data and convert to approximate percentage
        try:
            data = self.sensor()
            fan_rpm_values = []
            import re

            for line in data.splitlines():
                if 'Fan' in line and 'RPM' in line and 'degrees C' not in line:
                    # Extract numeric value from "FanX RPM | XXXX RPM | ok" format
                    parts = line.split('|')
                    if len(parts) >= 2:
                        rpm_part = parts[1].strip()
                        # Use regex to extract RPM value
                        rpm_match = re.search(r'(\d+)\s+RPM', rpm_part)
                        if rpm_match:
                            rpm_value = int(rpm_match.group(1))
                            fan_rpm_values.append(rpm_value)

            if fan_rpm_values:
                # Calculate average RPM
                avg_rpm = sum(fan_rpm_values) / len(fan_rpm_values)

                # Based on calibration: 20% setting results in 4800 RPM
                # Therefore, 100% would theoretically be 24000 RPM (4800 * 5)
                # This seems high for typical server fans, but we'll use the calibrated ratio
                # When 20% = 4800 RPM, the percentage = (current_rpm / 4800) * 20
                calibrated_rpm_at_20_percent = 4800
                calibrated_percentage = 20  # This is the known setting

                # Calculate the theoretical max RPM based on the calibration
                theoretical_max_rpm = calibrated_rpm_at_20_percent * (100 // calibrated_percentage)  # 100/20 = 5

                # Calculate the current percentage
                estimated_percentage = min(100, int((avg_rpm / theoretical_max_rpm) * 100))

                # Round to nearest 5 to match typical percentage steps
                estimated_percentage = round(estimated_percentage / 5) * 5
                return min(100, estimated_percentage)
        except Exception as e:
            logger.warning(f'解析传感器数据获取风扇RPM失败: {e}')

        return -1  # Return -1 if unable to determine

    def switch_fan_mode(self, auto: bool):
        """
        switch the fan mode
        :param auto:
        :return:
        """
        manual_cmd = 'raw 0x30 0x30 0x01 0x00'
        auto_cmd = 'raw 0x30 0x30 0x01 0x01'
        return self.run_cmd(cmd=auto_cmd) if auto else self.run_cmd(cmd=manual_cmd)

    def set_fan_speed(self, speed: int):
        """
        set fan speed
        :param speed:
        :return:
        """
        if speed < 10 or speed > 100:
            raise ValueError(
                'speed must be between 10 and 100'
            )

        self.switch_fan_mode(auto=False)
        base_cmd = 'raw 0x30 0x30 0x02 0xff'
        return self.run_cmd(cmd=f'{base_cmd} {hex(speed)}')