Execution multithread
This commit is contained in:
12
config.yaml
12
config.yaml
@@ -82,18 +82,6 @@ devices:
|
||||
HA_device_class: "connectivity"
|
||||
HA_platform: "binary_sensor"
|
||||
|
||||
# Example of how to add another device:
|
||||
# another_device:
|
||||
# ip: "192.168.10.3"
|
||||
# snmp_community: "public"
|
||||
# oids:
|
||||
# - name: "cpu_usage"
|
||||
# oid: ".1.3.6.1.4.1.14988.1.1.3.14.0" # MikroTik CPU usage
|
||||
# type: "int"
|
||||
# HA_device_class: "power_factor"
|
||||
# HA_platform: "sensor"
|
||||
# HA_unit: "%"
|
||||
|
||||
# OID Configuration Reference:
|
||||
# - name: Unique identifier for this metric (used in MQTT topics and Home Assistant)
|
||||
# - oid: SNMP Object Identifier
|
||||
|
||||
165
snmp2mqtt.py
165
snmp2mqtt.py
@@ -11,12 +11,83 @@ import yaml
|
||||
import argparse
|
||||
import sys
|
||||
import os
|
||||
import threading
|
||||
import signal
|
||||
import time
|
||||
|
||||
logging.basicConfig(
|
||||
format='(%(levelname)s) %(message)s',
|
||||
format='(%(levelname)s) [%(threadName)s] %(message)s',
|
||||
level=logging.DEBUG
|
||||
)
|
||||
|
||||
# Global shutdown flag
|
||||
shutdown_event = threading.Event()
|
||||
|
||||
|
||||
class DeviceMonitorThread(threading.Thread):
|
||||
"""Thread class for monitoring a single device"""
|
||||
|
||||
def __init__(self, device_name, device_config, mqtt_config, sleep_interval=2):
|
||||
super().__init__(name=f"Device-{device_name}")
|
||||
self.device_name = device_name
|
||||
self.device_config = device_config
|
||||
self.mqtt_config = mqtt_config.copy()
|
||||
self.sleep_interval = sleep_interval
|
||||
self.daemon = True # Dies when main thread dies
|
||||
|
||||
# Create unique client ID for this device
|
||||
self.mqtt_config['client_id'] = f"snmp-mqtt-{device_name}-{random.randint(0, 1000)}"
|
||||
|
||||
# Create device request object
|
||||
self.req = {
|
||||
"device_name": device_name,
|
||||
"ip": device_config["ip"],
|
||||
"snmp_community": device_config["snmp_community"],
|
||||
"oids": device_config["oids"]
|
||||
}
|
||||
|
||||
def run(self):
|
||||
"""Main thread execution"""
|
||||
logging.info(f"Starting monitoring thread for device: {self.device_name} ({self.device_config['ip']})")
|
||||
|
||||
try:
|
||||
# Setup MQTT connection and Home Assistant config
|
||||
ha_config = ha_create_config(self.req)
|
||||
client = connect_mqtt(self.mqtt_config)
|
||||
client.loop_start()
|
||||
|
||||
config_topic = f"homeassistant/device/{ha_config['dev']['ids']}/config"
|
||||
state_topic = ha_config['state_topic']
|
||||
|
||||
logging.info(f"[{self.device_name}] MQTT client connected, starting monitoring loop")
|
||||
|
||||
while not shutdown_event.is_set():
|
||||
try:
|
||||
# Publish Home Assistant configuration
|
||||
publish(config_topic, client, ha_config, True, 0)
|
||||
logging.debug(f"[{self.device_name}] Published config to {config_topic}")
|
||||
|
||||
# Get SNMP data and publish state
|
||||
state = asyncio.run(get_snmp(self.req))
|
||||
publish(state_topic, client, state, False, 0)
|
||||
logging.debug(f"[{self.device_name}] Published state to {state_topic}: {json.dumps(state)}")
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"[{self.device_name}] Error in monitoring loop: {e}")
|
||||
|
||||
# Wait for next iteration or shutdown signal
|
||||
shutdown_event.wait(timeout=self.sleep_interval)
|
||||
|
||||
# Cleanup
|
||||
client.loop_stop()
|
||||
client.disconnect()
|
||||
logging.info(f"[{self.device_name}] Monitoring thread stopped gracefully")
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"[{self.device_name}] Fatal error in monitoring thread: {e}")
|
||||
|
||||
logging.info(f"[{self.device_name}] Thread {self.name} finished")
|
||||
|
||||
|
||||
def parse_arguments():
|
||||
"""Parse command line arguments"""
|
||||
@@ -180,63 +251,55 @@ def ha_create_config(req):
|
||||
return ha_config
|
||||
|
||||
|
||||
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(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(f"Error getting SNMP data for {device_name}: {e}")
|
||||
pass
|
||||
sleep(sleep_interval)
|
||||
def signal_handler(signum, frame):
|
||||
"""Handle shutdown signals gracefully"""
|
||||
logging.info(f"Received signal {signum}, initiating graceful shutdown...")
|
||||
shutdown_event.set()
|
||||
|
||||
|
||||
def process_devices(config):
|
||||
"""Process multiple devices from configuration"""
|
||||
"""Process multiple devices using threading"""
|
||||
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")
|
||||
# Setup signal handlers for graceful shutdown
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
signal.signal(signal.SIGTERM, signal_handler)
|
||||
|
||||
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)
|
||||
device_threads = []
|
||||
|
||||
try:
|
||||
logging.info(f"Starting monitoring for {len(config['devices'])} device(s)")
|
||||
|
||||
# Create and start a thread for each device
|
||||
for device_name, device_config in config['devices'].items():
|
||||
thread = DeviceMonitorThread(
|
||||
device_name=device_name,
|
||||
device_config=device_config,
|
||||
mqtt_config=mqtt_config,
|
||||
sleep_interval=sleep_interval
|
||||
)
|
||||
device_threads.append(thread)
|
||||
thread.start()
|
||||
logging.info(f"Started thread for device: {device_name}")
|
||||
|
||||
# Wait for all threads to complete or shutdown signal
|
||||
while any(thread.is_alive() for thread in device_threads) and not shutdown_event.is_set():
|
||||
time.sleep(0.5)
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Error in process_devices: {e}")
|
||||
shutdown_event.set()
|
||||
|
||||
# Wait for all threads to finish
|
||||
logging.info("Waiting for all monitoring threads to finish...")
|
||||
for thread in device_threads:
|
||||
if thread.is_alive():
|
||||
thread.join(timeout=5.0) # Wait max 5 seconds per thread
|
||||
if thread.is_alive():
|
||||
logging.warning(f"Thread {thread.name} did not stop gracefully")
|
||||
|
||||
logging.info("All monitoring threads have finished")
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
Reference in New Issue
Block a user