diff --git a/cmk/plugins/raritan/__init__.py b/cmk/plugins/raritan/__init__.py index 0f803f40806..284e38bcca2 100644 --- a/cmk/plugins/raritan/__init__.py +++ b/cmk/plugins/raritan/__init__.py @@ -12,3 +12,14 @@ # (c) 2024 SWR # @author Frank Baier # +from cmk.plugins.raritan.lib import ( + InletSensorEnabledThresholds, + InletDeviceCapabilities, + InletPoleCapabilities, + RaritanData, + SensorValues, + Sensor, + PDU, + TYPE_MAPPING, + UNIT_MAPPING, +) diff --git a/cmk/plugins/raritan/agent_based/raritan_px2_residual_operating_current.py b/cmk/plugins/raritan/agent_based/raritan_px2_residual_operating_current.py index 2bfe29f59af..2407f3a68ea 100644 --- a/cmk/plugins/raritan/agent_based/raritan_px2_residual_operating_current.py +++ b/cmk/plugins/raritan/agent_based/raritan_px2_residual_operating_current.py @@ -16,7 +16,6 @@ from collections.abc import Callable, Sequence -from dataclasses import dataclass, field from typing import ( NotRequired, TypedDict, @@ -41,65 +40,18 @@ State, StringByteTable, ) +from cmk.plugins.raritan import ( + InletSensorEnabledThresholds, + InletDeviceCapabilities, + InletPoleCapabilities, + RaritanData, + SensorValues, + Sensor, + PDU, + TYPE_MAPPING, + UNIT_MAPPING, +) -TYPE_MAPPING = { - "1": ("current", "RMS"), - "2": ("peak", "Peak"), - "3": ("unbalanced", "Unbalanced"), - "4": ("voltage", "RMS"), - "5": ("power", "Active"), - "6": ("appower", "Apparent"), - # power factor is defined as the ratio of the real power flowing - # to the load to the apparent power - "7": ("power_factor", "Power Factor"), - "8": ("energy", "Active"), - "9": ("energy", "Apparent"), - "10": ("temp", ""), - "11": ("humidity", ""), - "12": ("airflow", ""), - "13": ("pressure_pa", "Air"), - "14": ("binary", "On/Off"), - "15": ("binary", "Trip"), - "16": ("binary", "Vibration"), - "17": ("binary", "Water Detector"), - "18": ("binary", "Smoke Detector"), - "19": ("binary", ""), - "20": ("binary", "Contact"), - "21": ("fanspeed", ""), - "26": ("residual_current", "Residual Current"), - "30": ("", "Other"), - "31": ("", "None"), -} - -UNIT_MAPPING = { - "-1": "", - "0": " Other", - "1": " V", - "2": " A", - "3": " W", - "4": " VA", - "5": " Wh", - "6": " VAh", - # for dev_unit in check_temperature - "7": "c", - "8": " hz", - "9": "%", - "10": " m/s", - "11": " Pa", - # 1 psi = 6894,757293168 Pa - "12": " psi", - "13": " g", - # for dev_unit in check_temperature - "14": "f", - "15": " ft", - "16": " inch", - "17": " cm", - "18": " m", - "19": " RPM", -} - - -RESIDUAL_BITMASK = 0b01000000 _RENDER_FUNCTION_AND_UNIT: dict[str, tuple[Callable | None, str]] = { "%": ( @@ -119,46 +71,6 @@ class Params(TypedDict): residual_levels: NotRequired[NoLevelsT | FixedLevelsT] -@dataclass(frozen=True, kw_only=True) -class PDU: - pdu_index: str - label: str - name: str - plug: str - pole_count: str - rated_voltage: str - rated_current: str - rated_frequency: str - rated_va: str - plug_descriptor: str - enable_state: str - device_capabilities: list[int] - pole_capabilities: list[int] - - -@dataclass(frozen=True, kw_only=True) -class SensorValues: - sensor_value: float - sensor_upper_crit: float - sensor_upper_warn: float - - -@dataclass(frozen=True, kw_only=True) -class Sensor: - availability: str - sensor_name: str - sensor_type: str - sensor_values: SensorValues - sensor_unit: str - sensor_thresholds: list[int] - - -@dataclass(kw_only=True) -class RaritanData: - pdu: PDU - sensors: dict[str, dict[str, Sensor]] = field(default_factory=dict) - - def _parse_pdu_data(raw_pdu_data: StringByteTable) -> PDU: pdu_data = raw_pdu_data[0] @@ -172,8 +84,8 @@ def _parse_pdu_data(raw_pdu_data: StringByteTable) -> PDU: rated_current=str(pdu_data[6]), rated_frequency=str(pdu_data[7]), rated_va=str(pdu_data[8]), - device_capabilities=pdu_data[9] if isinstance(pdu_data[9], list) else [int(pdu_data[9])], - pole_capabilities=pdu_data[10] if isinstance(pdu_data[10], list) else [int(pdu_data[10])], + device_capabilities=InletDeviceCapabilities.from_snmp_bits(pdu_data[9]), + pole_capabilities=InletPoleCapabilities.from_snmp_bits(pdu_data[10]), plug_descriptor=str(pdu_data[11]), enable_state=str(pdu_data[12]), ) @@ -235,9 +147,7 @@ def _parse_sensor_data(sensor_data: StringByteTable) -> dict[str, dict[str, Sens decimal_digits_int, ), sensor_unit=sensor_unit.strip(), - sensor_thresholds=enabled_thresholds - if isinstance(enabled_thresholds, list) - else [int(enabled_thresholds)], + sensor_thresholds=InletSensorEnabledThresholds.from_snmp_bits(enabled_thresholds), ) return sensors @@ -250,9 +160,9 @@ def parse_raritan_inlet_sensors( return None section = RaritanData(pdu=_parse_pdu_data(raw_pdu_data)) - if int(section.pdu.device_capabilities[3]) & RESIDUAL_BITMASK: + if section.pdu.device_capabilities.residualCurrent: section.sensors.update(_parse_sensor_data(string_table[1])) - if int(section.pdu.pole_capabilities[3]) & RESIDUAL_BITMASK: + if section.pdu.pole_capabilities.residualCurrent: section.sensors.update(_parse_sensor_data(string_table[2])) return section @@ -346,16 +256,19 @@ def _create_levels( params: Params, sensor_data: Sensor, ) -> NoLevelsT | FixedLevelsT: - thresholds = sensor_data.sensor_thresholds[0] - has_warn = thresholds & 0b00100000 and sensor_data.sensor_values.sensor_upper_warn != 0 - has_crit = thresholds & 0b00010000 and sensor_data.sensor_values.sensor_upper_crit != 0 - - if has_warn or has_crit: - levels_upper_warn = sensor_data.sensor_values.sensor_upper_warn if has_warn else None - levels_upper_crit = sensor_data.sensor_values.sensor_upper_crit if has_crit else None - return ("fixed", (levels_upper_warn, levels_upper_crit)) - return params.get("residual_levels", ("no_levels", None)) + has_warn = sensor_data.sensor_thresholds.upperWarning and sensor_data.sensor_values.sensor_upper_warn != 0 + has_crit = sensor_data.sensor_thresholds.upperCritical and sensor_data.sensor_values.sensor_upper_crit != 0 + + if params.get("residual_levels", False): + levels_upper = params.get("residual_levels", NoLevelsT) + elif has_warn or has_crit: + levels_upper_warn = sensor_data.sensor_values.sensor_upper_warn if has_warn else has_crit + levels_upper_crit = sensor_data.sensor_values.sensor_upper_crit if has_crit else has_warn + levels_upper = ("fixed", (levels_upper_warn, levels_upper_crit)) + else: + levels_upper = NoLevelsT + return levels_upper def _check_missing_levels( diff --git a/cmk/plugins/raritan/lib.py b/cmk/plugins/raritan/lib.py index 03c613d93c6..f11b9709474 100644 --- a/cmk/plugins/raritan/lib.py +++ b/cmk/plugins/raritan/lib.py @@ -2,7 +2,244 @@ # Copyright (C) 2023 Checkmk GmbH - License: GNU General Public License v2 # This file is part of Checkmk (https://checkmk.com). It is subject to the terms and # conditions defined in the file COPYING, which is part of this source code package. - -from cmk.agent_based.v2 import equals +from collections.abc import Sequence +from pydantic import BaseModel, ConfigDict, Field +from typing import ClassVar, Self +from cmk.agent_based.v2 import equals, StringByteTable DETECT_RARITAN = equals(".1.3.6.1.2.1.1.2.0", ".1.3.6.1.4.1.13742.6") + +TYPE_MAPPING = { + "1": ("current", "RMS"), + "2": ("peak", "Peak"), + "3": ("unbalanced", "Unbalanced"), + "4": ("voltage", "RMS"), + "5": ("power", "Active"), + "6": ("appower", "Apparent"), + # power factor is defined as the ratio of the real power flowing + # to the load to the apparent power + "7": ("power_factor", "Power Factor"), + "8": ("energy", "Active"), + "9": ("energy", "Apparent"), + "10": ("temp", ""), + "11": ("humidity", ""), + "12": ("airflow", ""), + "13": ("pressure_pa", "Air"), + "14": ("binary", "On/Off"), + "15": ("binary", "Trip"), + "16": ("binary", "Vibration"), + "17": ("binary", "Water Detector"), + "18": ("binary", "Smoke Detector"), + "19": ("binary", ""), + "20": ("binary", "Contact"), + "21": ("fanspeed", ""), + "26": ("residual_current", "Residual Current"), + "30": ("", "Other"), + "31": ("", "None"), +} + +UNIT_MAPPING = { + "-1": "", + "0": " Other", + "1": " V", + "2": " A", + "3": " W", + "4": " VA", + "5": " Wh", + "6": " VAh", + # for dev_unit in check_temperature + "7": "c", + "8": " hz", + "9": "%", + "10": " m/s", + "11": " Pa", + # 1 psi = 6894,757293168 Pa + "12": " psi", + "13": " g", + # for dev_unit in check_temperature + "14": "f", + "15": " ft", + "16": " inch", + "17": " cm", + "18": " m", + "19": " RPM", +} + + +class SnmpBitsModel(BaseModel): + model_config = ConfigDict(frozen=True) + + BIT_MAPPING: ClassVar[dict[int, str]] + + @classmethod + def from_snmp_bits( + cls, + value: bytes | bytearray | Sequence[int], + ) -> Self: + raw = value.encode("latin-1") if isinstance(value, str) else value + return cls.model_validate( + { + name: cls._bit_is_set(raw, bit_number) + for bit_number, name in cls.BIT_MAPPING.items() + } + ) + + @staticmethod + def _bit_is_set( + value: bytes | bytearray | Sequence[int], + bit_number: int, + ) -> bool: + byte_index, bit_index = divmod(bit_number, 8) + + if byte_index >= len(value): + return False + + return bool(value[byte_index] & (0x80 >> bit_index)) + + +class InletDeviceCapabilities(SnmpBitsModel): + BIT_MAPPING = { + 0: "rmsCurrent", + 1: "peakCurrent", + 2: "unbalancedCurrent", + 3: "rmsVoltage", + 4: "activePower", + 5: "apparentPower", + 6: "powerFactor", + 7: "activeEnergy", + 8: "apparentEnergy", + 21: "surgeProtectorStatus", + 22: "frequency", + 23: "phaseAngle", + 25: "residualCurrent", + 26: "rcmState", + 28: "reactivePower", + 31: "powerQuality", + 34: "displacementPowerFactor", + 35: "residualDcCurrent", + } + + rmsCurrent: bool = False + peakCurrent: bool = False + unbalancedCurrent: bool = False + rmsVoltage: bool = False + activePower: bool = False + apparentPower: bool = False + powerFactor: bool = False + activeEnergy: bool = False + apparentEnergy: bool = False + surgeProtectorStatus: bool = False + frequency: bool = False + phaseAngle: bool = False + residualCurrent: bool = False + rcmState: bool = False + reactivePower: bool = False + powerQuality: bool = False + displacementPowerFactor: bool = False + residualDcCurrent: bool = False + + +class InletPoleCapabilities(SnmpBitsModel): + BIT_MAPPING = { + 0: "rmsCurrent", + 1: "peakCurrent", + 3: "rmsVoltage", + 4: "activePower", + 5: "apparentPower", + 6: "powerFactor", + 7: "activeEnergy", + 8: "apparentEnergy", + 23: "phaseAngle", + 24: "rmsVoltageLN", + 25: "residualCurrent", + 26: "rcmState", + 28: "reactivePower", + 34: "displacementPowerFactor", + 35: "residualDcCurrent", + } + + rmsCurrent: bool = False + peakCurrent: bool = False + rmsVoltage: bool = False + activePower: bool = False + apparentPower: bool = False + powerFactor: bool = False + activeEnergy: bool = False + apparentEnergy: bool = False + phaseAngle: bool = False + rmsVoltageLN: bool = False + residualCurrent: bool = False + rcmState: bool = False + reactivePower: bool = False + displacementPowerFactor: bool = False + residualDcCurrent: bool = False + + +class InletSensorEnabledThresholds(SnmpBitsModel): + BIT_MAPPING = { + 0: "lowerCritical", + 1: "lowerWarning", + 2: "upperWarning", + 3: "upperCritical", + } + + lowerCritical: bool = False + lowerWarning: bool = False + upperWarning: bool = False + upperCritical: bool = False + + + +class PDU(BaseModel): + model_config = ConfigDict( + frozen=True, + extra="forbid", + ) + + pdu_index: str + label: str + name: str + plug: str + pole_count: str + rated_voltage: str + rated_current: str + rated_frequency: str + rated_va: str + plug_descriptor: str + enable_state: str + device_capabilities: InletDeviceCapabilities + pole_capabilities: InletPoleCapabilities + + +class SensorValues(BaseModel): + model_config = ConfigDict( + frozen=True, + extra="forbid", + ) + + sensor_value: float + sensor_upper_crit: float + sensor_upper_warn: float + + +class Sensor(BaseModel): + model_config = ConfigDict( + frozen=True, + extra="forbid", + ) + + availability: str + sensor_name: str + sensor_type: str + sensor_values: SensorValues + sensor_unit: str + sensor_thresholds: InletSensorEnabledThresholds + + +class RaritanData(BaseModel): + model_config = ConfigDict( + extra="forbid", + ) + + pdu: PDU + sensors: dict[str, dict[str, Sensor]] = Field(default_factory=dict)