AntoineVe/snmp2mqtt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Ajout d'un fichier de configuraiton séparé

Browse Source
This commit is contained in:
Antoine Van Elstraete
2025-08-24 14:32:06 +02:00
parent 819ff2ed97
commit ca4ad8ed06
3 changed files with 254 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
snmp2mqtt.py
View File

@@ -7,6 +7,10 @@ import random
from paho.mqtt import client as mqtt_client
import json
from time import sleep
import yaml
import argparse
import sys
import os
logging.basicConfig(
format='(%(levelname)s) %(message)s',
@@ -14,6 +18,77 @@ logging.basicConfig(
)
def parse_arguments():
"""Parse command line arguments"""
parser = argparse.ArgumentParser(description='SNMP to MQTT bridge for Home Assistant')
parser.add_argument('--config', '-c', required=True,
help='Path to YAML configuration file')
return parser.parse_args()
def load_config(config_path):
"""Load and validate YAML configuration file"""
if not os.path.exists(config_path):
logging.error(f"Configuration file not found: {config_path}")
sys.exit(1)
try:
with open(config_path, 'r') as file:
config = yaml.safe_load(file)
except yaml.YAMLError as e:
logging.error(f"Error parsing YAML configuration: {e}")
sys.exit(1)
except Exception as e:
logging.error(f"Error reading configuration file: {e}")
sys.exit(1)
# Validate required configuration sections
if 'mqtt' not in config:
logging.error("Missing 'mqtt' section in configuration")
sys.exit(1)
if 'devices' not in config:
logging.error("Missing 'devices' section in configuration")
sys.exit(1)
# Validate MQTT configuration
required_mqtt_fields = ['broker', 'port', 'user', 'password']
for field in required_mqtt_fields:
if field not in config['mqtt']:
logging.error(f"Missing required MQTT field: {field}")
sys.exit(1)
# Validate device configurations
for device_name, device_config in config['devices'].items():
required_device_fields = ['ip', 'snmp_community', 'oids']
for field in required_device_fields:
if field not in device_config:
logging.error(f"Missing required field '{field}' in device '{device_name}'")
sys.exit(1)
# Validate OID configurations
for oid in device_config['oids']:
required_oid_fields = ['name', 'oid', 'type', 'HA_device_class', 'HA_platform']
for field in required_oid_fields:
if field not in oid:
logging.error(f"Missing required OID field '{field}' in device '{device_name}'")
sys.exit(1)
# Convert type string to actual Python type
if oid['type'] == 'int':
oid['type'] = int
elif oid['type'] == 'bool':
oid['type'] = bool
elif oid['type'] == 'str':
oid['type'] = str
else:
logging.error(f"Unsupported type '{oid['type']}' for OID '{oid['name']}' in device '{device_name}'")
sys.exit(1)
logging.info(f"Configuration loaded successfully from {config_path}")
return config
def connect_mqtt(mqtt_config):
def on_connect(client, userdata, flags, rc):
if rc == 0:
@@ -105,105 +180,81 @@ def ha_create_config(req):
return ha_config
def send_to_mqtt():
def send_to_mqtt(device_name, device_config, mqtt_config, sleep_interval=2):
"""Send SNMP data to MQTT for a single device"""
# Create device request object
req = {
"device_name": device_name,
"ip": device_config["ip"],
"snmp_community": device_config["snmp_community"],
"oids": device_config["oids"]
}
config = ha_create_config(req)
client = connect_mqtt(mqtt_config)
client.loop_start()
config_topic = f"homeassistant/device/{config['dev']['ids']}/config"
state_topic = config['state_topic']
while True:
try:
publish(config_topic, client, config, True, 0)
logging.info(f"{config_topic} -> {config}")
except Exception as e:
logging.error(e)
logging.error(f"Error publishing config for {device_name}: {e}")
pass
try:
state = asyncio.run(get_snmp(req))
publish(state_topic, client, state, False, 0)
logging.info(f"{state_topic} -> {state}")
except Exception as e:
logging.error(e)
logging.error(f"Error getting SNMP data for {device_name}: {e}")
pass
sleep(2)
sleep(sleep_interval)
req = {
"device_name": "mikrotik_hex",
"ip": "192.168.10.2",
"snmp_community": "public",
"oids": [
{"name": "stln_vpn_in",
"oid": ".1.3.6.1.2.1.2.2.1.10.12",
"type": int,
"HA_device_class": "data_size",
"HA_platform": "sensor",
"HA_unit": "bit"
},
{"name": "stlon_vpn_out",
"oid": ".1.3.6.1.2.1.2.2.1.16.12",
"type": int,
"HA_device_class": "data_size",
"HA_platform": "sensor",
"HA_unit": "bit"
},
{"name": "stln_vpn_status",
"oid": ".1.3.6.1.2.1.2.2.1.8.12",
"type": bool,
"HA_device_class": "connectivity",
"HA_platform": "binary_sensor",
},
{"name": "lan_bridge_in",
"oid": ".1.3.6.1.2.1.2.2.1.10.6",
"type": int,
"HA_device_class": "data_size",
"HA_platform": "sensor",
"HA_unit": "bit"
},
{"name": "lan_bridge_out",
"oid": ".1.3.6.1.2.1.2.2.1.16.6",
"type": int,
"HA_device_class": "data_size",
"HA_platform": "sensor",
"HA_unit": "bit"
},
{"name": "lan_bridge_status",
"oid": ".1.3.6.1.2.1.2.2.1.8.6",
"type": bool,
"HA_device_class": "connectivity",
"HA_platform": "binary_sensor",
},
{"name": "starlink_in",
"oid": ".1.3.6.1.2.1.2.2.1.10.1",
"type": int,
"HA_device_class": "data_size",
"HA_platform": "sensor",
"HA_unit": "bit"
},
{"name": "starlink_out",
"oid": ".1.3.6.1.2.1.2.2.1.16.1",
"type": int,
"HA_device_class": "data_size",
"HA_platform": "sensor",
"HA_unit": "bit"
},
{"name": "starlink_status",
"oid": ".1.3.6.1.2.1.2.2.1.8.1",
"type": bool,
"HA_device_class": "connectivity",
"HA_platform": "binary_sensor",
}
]
}
mqtt_config = {
"broker": "192.168.10.202",
"port": 1883,
"client_id": f"snmp-mqtt-{random.randint(0, 1000)}",
"user": "snmp2mqtt",
"password": "snmp_2_MQTT"
}
def process_devices(config):
"""Process multiple devices from configuration"""
mqtt_config = config['mqtt'].copy()
mqtt_config['client_id'] = f"snmp-mqtt-{random.randint(0, 1000)}"
# Get sleep interval from config or use default
sleep_interval = config.get('sleep_interval', 2)
if len(config['devices']) == 1:
# Single device mode - run directly
device_name = list(config['devices'].keys())[0]
device_config = config['devices'][device_name]
logging.info(f"Starting monitoring for single device: {device_name}")
send_to_mqtt(device_name, device_config, mqtt_config, sleep_interval)
else:
# Multiple devices mode - would need threading/multiprocessing
# For now, let's process the first device and warn about others
logging.warning(f"Multiple devices detected ({len(config['devices'])}), but only processing the first one")
logging.warning("Multi-device support will require threading implementation")
device_name = list(config['devices'].keys())[0]
device_config = config['devices'][device_name]
logging.info(f"Starting monitoring for device: {device_name}")
send_to_mqtt(device_name, device_config, mqtt_config, sleep_interval)
send_to_mqtt()
def main():
"""Main entry point"""
args = parse_arguments()
config = load_config(args.config)
logging.info("Starting snmp2mqtt bridge...")
logging.info(f"Configured devices: {list(config['devices'].keys())}")
try:
process_devices(config)
except KeyboardInterrupt:
logging.info("Shutdown requested by user")
except Exception as e:
logging.error(f"Unexpected error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()