From 9e84d2aa75e9baebd3cd693c265984a69b5bdda1 Mon Sep 17 00:00:00 2001 From: Marcus Brown Date: Sun, 20 Oct 2024 17:52:41 +1100 Subject: [PATCH] Retrieve retained MQTT snapshot and disconnect after receiving - Modified the script to disconnect after receiving the retained snapshot message. - Simplified MQTT loop handling to ensure reliable retrieval of retained messages. --- dayglo_detector/dayglo_detector.py | 257 +---------------------------- 1 file changed, 3 insertions(+), 254 deletions(-) diff --git a/dayglo_detector/dayglo_detector.py b/dayglo_detector/dayglo_detector.py index bb05a1b..b51580f 100644 --- a/dayglo_detector/dayglo_detector.py +++ b/dayglo_detector/dayglo_detector.py @@ -64,6 +64,8 @@ def on_message(client, userdata, msg): print(f"Payload (first 100 bytes): {msg.payload[:100]}...") process_snapshot(msg.payload) initial_snapshot_processed = True + # Disconnect after receiving the retained message + client.disconnect() def process_snapshot(payload): if not payload: @@ -114,258 +116,5 @@ client.on_message = on_message client.enable_logger() connect_mqtt() -client.loop_start() - -# Keep the script running -while True: - time.sleep(1) - -import os -import time -import paho.mqtt.client as mqtt -import numpy as np -import cv2 -import base64 -import json -import tempfile - -# Configuration -MQTT_BROKER = os.environ.get('MQTT_BROKER', '10.59.221.172') -MQTT_PORT = int(os.environ.get('MQTT_PORT', '1883')) -MQTT_USERNAME = os.getenv('MQTT_USERNAME', 'your_username') -MQTT_PASSWORD = os.getenv('MQTT_PASSWORD', 'your_password') -MQTT_SNAPSHOT_TOPIC = "/frigate/patiocam/person/snapshot" -MQTT_TOPIC_PUBLISH = "homeassistant/sensor/dayglo_rating/state" -DISCOVERY_PREFIX = "homeassistant" - -# Default rating -last_rating = 0 - -# Default color thresholds for dayglo detection -LOWER_COLOR = np.array([20, 100, 100]) -UPPER_COLOR = np.array([40, 255, 255]) - -# Track if the initial snapshot has been processed -initial_snapshot_processed = False - -def on_connect(client, userdata, flags, rc): - if rc == 0: - print("Connected successfully to MQTT broker") - client.subscribe(MQTT_SNAPSHOT_TOPIC, qos=1) - print(f"Subscribed to topic: {MQTT_SNAPSHOT_TOPIC}") - publish_discovery_configurations() - # Publish initial rating of 0 - publish_rating(0) - # Attempt to request the retained snapshot message - client.publish(MQTT_SNAPSHOT_TOPIC, "", qos=1) - else: - print(f"Failed to connect, return code {rc}") - -def publish_discovery_configurations(): - rating_config = { - "name": "Dayglo Rating", - "state_topic": MQTT_TOPIC_PUBLISH, - "unit_of_measurement": "%", - "value_template": "{{ value_json.rating }}", - "icon": "mdi:brush", - "unique_id": "mqtt_dayglo_rating" - } - client.publish(f"{DISCOVERY_PREFIX}/sensor/dayglo_rating/config", json.dumps(rating_config), retain=True) - -def publish_rating(rating): - global last_rating - last_rating = rating - client.publish(MQTT_TOPIC_PUBLISH, json.dumps({"rating": rating})) - -def on_message(client, userdata, msg): - global initial_snapshot_processed - if msg.topic == MQTT_SNAPSHOT_TOPIC: - print("Snapshot received") - if len(msg.payload) == 0: - print("Received an empty payload, skipping processing.") - else: - print(f"Payload length: {len(msg.payload)} bytes") - print(f"Payload (first 100 bytes): {msg.payload[:100]}...") - process_snapshot(msg.payload) - initial_snapshot_processed = True - -def process_snapshot(payload): - if not payload: - print("Empty payload received, skipping processing.") - return - - print("Processing snapshot...") - try: - image_data = base64.b64decode(payload) - with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as temp_image_file: - temp_image_file.write(image_data) - temp_image_path = temp_image_file.name - - # Attempt to read the saved image with OpenCV - image = cv2.imread(temp_image_path) - - if image is not None: - rating = calculate_dayglo_rating(image) - print("Dayglo Rating calculated:", rating) - publish_rating(rating) - else: - print("Invalid image format or corrupted image received") - except Exception as e: - print(f"Error processing snapshot: {e}") - -def calculate_dayglo_rating(image): - print("Calculating dayglo rating...") - hsv_image = cv2.cvtColor(image, cv2.COLOR_BGR2HSV) - mask = cv2.inRange(hsv_image, LOWER_COLOR, UPPER_COLOR) - dayglo_pixels = cv2.countNonZero(mask) - total_pixels = image.shape[0] * image.shape[1] - rating = (dayglo_pixels / total_pixels) * 100 - return rating - -def connect_mqtt(): - while True: - try: - client.connect(MQTT_BROKER, MQTT_PORT, 60) - break - except Exception as e: - print(f"Connection failed: {e}. Retrying in 5 seconds...") - time.sleep(5) - -client = mqtt.Client(protocol=mqtt.MQTTv311) -client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD) -client.on_connect = on_connect -client.on_message = on_message -client.enable_logger() - -connect_mqtt() -client.loop_start() - -# Keep the script running -while True: - time.sleep(1) - -import os -import time -import paho.mqtt.client as mqtt -import numpy as np -import cv2 -import base64 -import json -import tempfile - -# Configuration -MQTT_BROKER = os.environ.get('MQTT_BROKER', '10.59.221.172') -MQTT_PORT = int(os.environ.get('MQTT_PORT', '1883')) -MQTT_USERNAME = os.getenv('MQTT_USERNAME', 'your_username') -MQTT_PASSWORD = os.getenv('MQTT_PASSWORD', 'your_password') -MQTT_SNAPSHOT_TOPIC = "/frigate/patiocam/person/snapshot" -MQTT_TOPIC_PUBLISH = "homeassistant/sensor/dayglo_rating/state" -DISCOVERY_PREFIX = "homeassistant" - -# Default rating -last_rating = 0 - -# Default color thresholds for dayglo detection -LOWER_COLOR = np.array([20, 100, 100]) -UPPER_COLOR = np.array([40, 255, 255]) - -# Track if the initial snapshot has been processed -initial_snapshot_processed = False - -def on_connect(client, userdata, flags, rc): - if rc == 0: - print("Connected successfully to MQTT broker") - client.subscribe(MQTT_SNAPSHOT_TOPIC, qos=1) - print(f"Subscribed to topic: {MQTT_SNAPSHOT_TOPIC}") - publish_discovery_configurations() - # Publish initial rating of 0 - publish_rating(0) - else: - print(f"Failed to connect, return code {rc}") - -def publish_discovery_configurations(): - rating_config = { - "name": "Dayglo Rating", - "state_topic": MQTT_TOPIC_PUBLISH, - "unit_of_measurement": "%", - "value_template": "{{ value_json.rating }}", - "icon": "mdi:brush", - "unique_id": "mqtt_dayglo_rating" - } - client.publish(f"{DISCOVERY_PREFIX}/sensor/dayglo_rating/config", json.dumps(rating_config), retain=True) - -def publish_rating(rating): - global last_rating - last_rating = rating - client.publish(MQTT_TOPIC_PUBLISH, json.dumps({"rating": rating})) - -def on_message(client, userdata, msg): - global initial_snapshot_processed - if msg.topic == MQTT_SNAPSHOT_TOPIC: - print("Snapshot received") - if len(msg.payload) == 0: - print("Received an empty payload, skipping processing.") - else: - print(f"Payload length: {len(msg.payload)} bytes") - print(f"Payload (first 100 bytes): {msg.payload[:100]}...") - process_snapshot(msg.payload) - initial_snapshot_processed = True - -def process_snapshot(payload): - if not payload: - print("Empty payload received, skipping processing.") - return - - print("Processing snapshot...") - try: - image_data = base64.b64decode(payload) - with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as temp_image_file: - temp_image_file.write(image_data) - temp_image_path = temp_image_file.name - - # Attempt to read the saved image with OpenCV - image = cv2.imread(temp_image_path) - - if image is not None: - rating = calculate_dayglo_rating(image) - print("Dayglo Rating calculated:", rating) - publish_rating(rating) - else: - print("Invalid image format or corrupted image received") - except Exception as e: - print(f"Error processing snapshot: {e}") - -def calculate_dayglo_rating(image): - print("Calculating dayglo rating...") - hsv_image = cv2.cvtColor(image, cv2.COLOR_BGR2HSV) - mask = cv2.inRange(hsv_image, LOWER_COLOR, UPPER_COLOR) - dayglo_pixels = cv2.countNonZero(mask) - total_pixels = image.shape[0] * image.shape[1] - rating = (dayglo_pixels / total_pixels) * 100 - return rating - -def connect_mqtt(): - while True: - try: - client.connect(MQTT_BROKER, MQTT_PORT, 60) - break - except Exception as e: - print(f"Connection failed: {e}. Retrying in 5 seconds...") - time.sleep(5) - -client = mqtt.Client(protocol=mqtt.MQTTv311) -client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD) -client.on_connect = on_connect -client.on_message = on_message -client.enable_logger() - -connect_mqtt() -client.loop_start() - -# Allow some time for the retained message to be received -time.sleep(5) - -# Keep the script running -while True: - time.sleep(1) +client.loop_forever()