wn/home-automation
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

brightness

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2025-10-31 15:31:35 +01:00
parent c3ec6e3fc4
commit 69e07056a1
4 changed files with 144 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
tools/sim_test_lampe.py
View File

@@ -1,10 +1,10 @@
#!/usr/bin/env python3
"""MQTT Simulator for test_lampe device.
"""MQTT Simulator for multiple test_lampe devices.
This simulator acts as a virtual light device that:
- Subscribes to vendor/test_lampe/set
- Maintains local state
- Publishes state changes to vendor/test_lampe/state (retained)
This simulator acts as virtual light devices that:
- Subscribe to vendor/test_lampe_*/set
- Maintain local state for each device
- Publish state changes to vendor/test_lampe_*/state (retained)
"""
import json
@@ -26,14 +26,20 @@ logger = logging.getLogger(__name__)
# Configuration
BROKER_HOST = os.environ.get("MQTT_HOST", "172.16.2.16")
BROKER_PORT = int(os.environ.get("MQTT_PORT", "1883"))
DEVICE_ID = "test_lampe_1"
SET_TOPIC = f"vendor/{DEVICE_ID}/set"
STATE_TOPIC = f"vendor/{DEVICE_ID}/state"
# Device state
device_state = {
"power": "off",
"brightness": 50
# Devices to simulate
DEVICES = ["test_lampe_1", "test_lampe_2"]
# Device states (one per device)
device_states = {
"test_lampe_1": {
"power": "off",
"brightness": 50
},
"test_lampe_2": {
"power": "off",
"brightness": 50
}
}
# Global client for signal handler
@@ -53,13 +59,16 @@ def on_connect(client, userdata, flags, rc, properties=None):
if rc == 0:
logger.info(f"Connected to MQTT broker {BROKER_HOST}:{BROKER_PORT}")
# Subscribe to SET topic
client.subscribe(SET_TOPIC, qos=1)
logger.info(f"Subscribed to {SET_TOPIC}")
# Subscribe to SET topics for all devices
for device_id in DEVICES:
set_topic = f"vendor/{device_id}/set"
client.subscribe(set_topic, qos=1)
logger.info(f"Subscribed to {set_topic}")
# Publish initial state (retained)
publish_state(client)
logger.info(f"Simulator started, initial state published: {device_state}")
# Publish initial states (retained)
for device_id in DEVICES:
publish_state(client, device_id)
logger.info(f"Simulator started for {device_id}, initial state: {device_states[device_id]}")
else:
logger.error(f"Connection failed with code {rc}")
@@ -72,53 +81,67 @@ def on_message(client, userdata, msg):
userdata: User data
msg: MQTT message
"""
global device_state
# Extract device_id from topic (vendor/test_lampe_X/set)
topic_parts = msg.topic.split('/')
if len(topic_parts) != 3 or topic_parts[0] != "vendor" or topic_parts[2] != "set":
logger.warning(f"Unexpected topic format: {msg.topic}")
return
device_id = topic_parts[1]
if device_id not in device_states:
logger.warning(f"Unknown device: {device_id}")
return
try:
payload = json.loads(msg.payload.decode())
logger.info(f"Received SET command: {payload}")
logger.info(f"[{device_id}] Received SET command: {payload}")
# Update device state
updated = False
device_state = device_states[device_id]
if "power" in payload:
old_power = device_state["power"]
device_state["power"] = payload["power"]
if old_power != device_state["power"]:
updated = True
logger.info(f"Power changed: {old_power} -> {device_state['power']}")
logger.info(f"[{device_id}] Power changed: {old_power} -> {device_state['power']}")
if "brightness" in payload:
old_brightness = device_state["brightness"]
device_state["brightness"] = int(payload["brightness"])
if old_brightness != device_state["brightness"]:
updated = True
logger.info(f"Brightness changed: {old_brightness} -> {device_state['brightness']}")
logger.info(f"[{device_id}] Brightness changed: {old_brightness} -> {device_state['brightness']}")
# Publish updated state if changed
if updated:
publish_state(client)
logger.info(f"Published new state: {device_state}")
publish_state(client, device_id)
logger.info(f"[{device_id}] Published new state: {device_state}")
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON in message: {e}")
logger.error(f"[{device_id}] Invalid JSON in message: {e}")
except Exception as e:
logger.error(f"Error processing message: {e}")
logger.error(f"[{device_id}] Error processing message: {e}")
def publish_state(client):
def publish_state(client, device_id):
"""Publish current device state to STATE topic.
Args:
client: MQTT client instance
device_id: Device identifier
"""
device_state = device_states[device_id]
state_topic = f"vendor/{device_id}/state"
state_json = json.dumps(device_state)
result = client.publish(STATE_TOPIC, state_json, qos=1, retain=True)
result = client.publish(state_topic, state_json, qos=1, retain=True)
if result.rc == mqtt.MQTT_ERR_SUCCESS:
logger.debug(f"Published state to {STATE_TOPIC}: {state_json}")
logger.debug(f"[{device_id}] Published state to {state_topic}: {state_json}")
else:
logger.error(f"Failed to publish state: {result.rc}")
logger.error(f"[{device_id}] Failed to publish state: {result.rc}")
def signal_handler(sig, frame):
@@ -131,11 +154,13 @@ def signal_handler(sig, frame):
logger.info(f"Received signal {sig}, shutting down...")
if client_global:
# Publish offline state before disconnecting
offline_state = device_state.copy()
offline_state["power"] = "off"
client_global.publish(STATE_TOPIC, json.dumps(offline_state), qos=1, retain=True)
logger.info("Published offline state")
# Publish offline state for all devices before disconnecting
for device_id in DEVICES:
offline_state = device_states[device_id].copy()
offline_state["power"] = "off"
state_topic = f"vendor/{device_id}/state"
client_global.publish(state_topic, json.dumps(offline_state), qos=1, retain=True)
logger.info(f"[{device_id}] Published offline state")
client_global.disconnect()
client_global.loop_stop()
@@ -153,7 +178,7 @@ def main():
# Create MQTT client
client = mqtt.Client(
client_id=f"simulator-{DEVICE_ID}",
client_id="simulator-test-lampes",
protocol=mqtt.MQTTv5,
callback_api_version=mqtt.CallbackAPIVersion.VERSION2
)