feat: Ajoute la transformation de valeur par opération mathématique

Cette modification introduit la possibilité d'appliquer une opération mathématique simple (addition, soustraction, multiplication, division) sur les valeurs SNMP récupérées.

L'utilisateur peut désormais définir une clé 'operation' dans la configuration d'un OID (par exemple, 'value / 1000') pour normaliser les données avant leur envoi à MQTT. La documentation dans `config.yaml` a été mise à jour avec un exemple.
This commit is contained in:
2025-11-06 17:05:20 +01:00
parent ac476df7ce
commit 7888999ec3
2 changed files with 61 additions and 2 deletions

View File

@@ -39,6 +39,17 @@ devices:
type: "bool"
HA_device_class: "connectivity"
HA_platform: "binary_sensor"
# Example of a temperature sensor that returns the value in millidegrees.
# The 'operation' key allows performing a simple calculation.
# The placeholder 'value' will be replaced by the SNMP value.
# - name: "temperature"
# oid: ".1.3.6.1.4.1.XXXX.1.1.1.5.1.3.1" # Example OID
# type: "int"
# operation: "value / 1000"
# HA_device_class: "temperature"
# HA_platform: "sensor"
# HA_unit: "°C"
# OID Configuration Reference:
# - name: Unique identifier for this metric (used in MQTT topics and Home Assistant)

View File

@@ -207,6 +207,46 @@ def publish(topic, client, data, retain, qos):
logging.error(f"Failed to send message to topic {topic}")
def apply_operation(value, operation_str):
"""
Applies a simple mathematical operation to a value.
e.g., operation_str = "value / 1000"
"""
if 'value' not in operation_str:
logging.error(f"Invalid operation string: 'value' placeholder missing in '{operation_str}'")
return value
expression = operation_str.replace('value', str(value))
try:
parts = expression.split()
if len(parts) != 3:
logging.error(f"Invalid operation format: '{operation_str}'. Expected 'value <operator> <operand>'")
return value
val = float(parts[0])
operator = parts[1]
operand = float(parts[2])
if operator == '+':
return val + operand
elif operator == '-':
return val - operand
elif operator == '*':
return val * operand
elif operator == '/':
if operand == 0:
logging.warning(f"Attempted division by zero in operation: {operation_str}")
return value
return val / operand
else:
logging.error(f"Unsupported operator: '{operator}' in '{operation_str}'")
return value
except (ValueError, IndexError) as e:
logging.error(f"Could not parse operation string: '{operation_str}'. Error: {e}")
return value
async def get_snmp(req):
"""Asynchronously retrieve SNMP data from device using new pysnmp API"""
data = {}
@@ -239,13 +279,21 @@ async def get_snmp(req):
else:
for varBind in varBinds:
logging.debug(f"{req['device_name']} {oid['name']} => {oid['type'](varBind[1])}")
# Cast to the right type
value = oid['type'](varBind[1])
# Apply operation if defined
if 'operation' in oid:
value = apply_operation(value, oid['operation'])
if oid['type'] == bool:
if bool(varBind[1]):
if bool(value):
data.update({oid["name"]: "ON"})
else:
data.update({oid["name"]: "OFF"})
else:
data.update({oid["name"]: oid["type"](varBind[1])})
data.update({oid["name"]: value})
except Exception as e:
logging.error(f"{req['device_name']} Exception getting OID {oid['oid']}: {e}")
continue