RC Stable7 7.1.015

Classes/OTA.py

@@ -133,7 +133,7 @@ class OTAManagement(object):
 
     def __init__(
         self,
-        zigbee_communitation,
+        zigbee_communication,
         PluginConf,
         DeviceConf,
         adminWidgets,
@@ -150,7 +150,7 @@ def __init__(
         ):
 
         # Pointers to external objects
-        self.zigbee_communication = zigbee_communitation
+        self.zigbee_communication = zigbee_communication
         self.HB = 0
         self.ListOfDevices = ListOfDevices  # Point to the Global ListOfDevices
         self.IEEE2NWK = IEEE2NWK  # Point to the List of IEEE to NWKID
@@ -702,26 +702,26 @@ def ota_upgrade_end_response(self, sqn, dest_addr, dest_ep, intMsgImageVersion,
     #
     # UPGRADE_END_RESPONSE 	0x0504
     # u32UpgradeTime is the UTC time, in seconds, at which the client should upgrade the running image with the downloaded image
-
     # u32CurrentTime is the current UTC time, in seconds, on the server.
-    _UpgradeTime = 0x00
+
     EPOCTime = datetime(2000, 1, 1)
     UTCTime = int((datetime.now() - EPOCTime).total_seconds())
+    _UpgradeTime = UTCTime + 10  # 10 seconds delay
 
     _FileVersion = intMsgImageVersion
     _ImageType = image_type
     _ManufacturerCode = intMsgManufCode
 
-    datas = "%02x" % ADDRESS_MODE["short"] + dest_addr + ZIGATE_EP + dest_ep
-    datas += "%08x" % _UpgradeTime
-    datas += "%08x" % 0x00
-    datas += "%08x" % _FileVersion
-    datas += "%04x" % _ImageType
-    datas += "%04x" % _ManufacturerCode
-
     if "ControllerInRawMode" in self.pluginconf.pluginConf and self.pluginconf.pluginConf["ControllerInRawMode"]:
         zcl_raw_ota_upgrade_end_response(self, sqn, dest_addr, ZIGATE_EP, dest_ep, "%04x" % _ManufacturerCode, "%04x" % _ImageType, "%08x" % _FileVersion, "%08x" %UTCTime, "%08x" % _UpgradeTime)
     else:
+        datas = "%02x" % ADDRESS_MODE["short"] + dest_addr + ZIGATE_EP + dest_ep
+        datas += "%08x" % _UpgradeTime
+        datas += "%08x" % 0x00
+        datas += "%08x" % _FileVersion
+        datas += "%04x" % _ImageType
+        datas += "%04x" % _ManufacturerCode
+
         self.ControllerLink.sendData("0504", datas, ackIsDisabled=False, NwkId=dest_addr)
 
     logging( self, "Log", "ota_management - sending Upgrade End Response, for %s Version: 0x%08X Type: 0x%04x, Manuf: 0x%04X" % (dest_addr, _FileVersion, _ImageType, _ManufacturerCode), )
@@ -1344,20 +1344,23 @@ def update_firmware_health(self, MsgSrcAddr, completion):
 
 
 def start_upgrade_infos(self, MsgSrcAddr, intMsgImageType, intMsgManufCode, MsgFileOffset, MsgMaxDataSize):  # OK 24/10/2020
+    """Start the firmware upgrade process for a device."""
 
+    # Retrieve the image entry for the requested image type
     entry = retrieve_image(self, intMsgImageType)
     if entry is None:
         logging(self, "Error", "start_upgrade_infos: No Firmware available to satify this request by %s !!!" % MsgSrcAddr)
         return
     brand, ota_image_file = entry
 
     available_image = self.ListOfImages["Brands"][brand][ota_image_file]
+
+    # Populate `ListInUpdate` with image details
     self.ListInUpdate["intSize"] = available_image["intSize"]
     self.ListInUpdate["ImageVersion"] = available_image["intImageVersion"]
     self.ListInUpdate["Process"] = available_image["Process"]
     self.ListInUpdate["Decoded Header"] = available_image["Decoded Header"]
     self.ListInUpdate["OtaImage"] = available_image["OtaImage"]
-
     self.ListInUpdate["ImageType"] = "%04x" % intMsgImageType
     self.ListInUpdate["intImageType"] = intMsgImageType
     self.ListInUpdate["NwkId"] = MsgSrcAddr
@@ -1368,28 +1371,33 @@ def start_upgrade_infos(self, MsgSrcAddr, intMsgImageType, intMsgManufCode, MsgF
     self.ListInUpdate["LastBlockSent"] = 0
     self.ListInUpdate["StartTime"] = time.time()
 
+    # Initialize or reset the "Firmware Update" section in PluginHealth
     if "Firmware Update" not in self.PluginHealth:
         self.PluginHealth["Firmware Update"] = {}
     if "Firmware Update" in self.PluginHealth:
         self.PluginHealth["Firmware Update"] = {}
     if self.PluginHealth["Firmware Update"] is None:
         self.PluginHealth["Firmware Update"] = {}
 
+    # Initialize or reset the "Firmware Update" section in PluginHealth
     self.PluginHealth["Firmware Update"]["Progress"] = "0%"
     self.PluginHealth["Firmware Update"]["Device"] = MsgSrcAddr
 
+    # Retrieve device name from the IEEE address
     _ieee = self.ListOfDevices[MsgSrcAddr]["IEEE"]
-
     _name = next((self.Devices[x].Name for x in self.Devices if self.Devices[x].DeviceID == _ieee), None)
 
-    _durhh, _durmm, _durss = convert_time(self.ListInUpdate["intSize"] // MsgMaxDataSize)
+    # Estimate upload time
+    estimated_time_for_upload = ( self.ListInUpdate["intSize"] // MsgMaxDataSize )
+    if self.zigbee_communication == "zigpy":
+        estimated_time_for_upload //= 7
+
+    # Convert estimated time into hours, minutes, and seconds
+    _durhh, _durmm, _durss = convert_time(estimated_time_for_upload)
+
+    # Generate notification text
     _textmsg = "Firmware update started for Device: %s with %s - Estimated Time: %s H %s min %s sec " % (
-        _name,
-        self.ListInUpdate["FileName"],
-        _durhh,
-        _durmm,
-        _durss,
-    )
+        _name, self.ListInUpdate["FileName"], _durhh, _durmm, _durss, )
     self.adminWidgets.updateNotificationWidget(self.Devices, _textmsg)
 
 

Modules/command.py

@@ -464,9 +464,12 @@ def handle_command_off(self,Devices, DeviceID, Unit, Level, Nwkid, EPout, Device
         self.log.logging("Command", "Debug", "handle_command_off : Disable Window Cover Calibration")
         tuya_window_cover_calibration(self, Nwkid, "01")
 
-    elif DeviceType == "Switch" and ts0601_extract_data_point_infos( self, model_name):
+    elif (
+        DeviceType == "Switch"
+        and not get_deviceconf_parameter_value(self, model_name, "StandardZigbeeCommand", return_default=False)
+        and ts0601_extract_data_point_infos( self, model_name)
+        ):
         ts0601_actuator(self, Nwkid, "switch", 0)
-
     else:
         # Remaining Slider widget
         _off_command_default(self, Nwkid, EPout, profalux, model_name)
@@ -653,7 +656,11 @@ def handle_command_on(self,Devices, DeviceID, Unit, Level, Nwkid, EPout, DeviceT
         self.log.logging("Command", "Debug", "mgtCommand : Enable Window Cover Calibration")
         tuya_window_cover_calibration(self, Nwkid, "00")
 
-    elif DeviceType == "Switch" and ts0601_extract_data_point_infos( self, model_name):
+    elif (
+        DeviceType == "Switch"
+        and not get_deviceconf_parameter_value(self, model_name, "StandardZigbeeCommand", return_default=False)
+        and ts0601_extract_data_point_infos( self, model_name)
+        ):
         ts0601_actuator(self, Nwkid, "switch", 1)
 
     elif profalux:

Modules/matomo_request.py

@@ -1,11 +1,23 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+#
+# Implementation of Zigbee for Domoticz plugin.
+#
+# This file is part of Zigbee for Domoticz plugin. https://github.com/zigbeefordomoticz/Domoticz-Zigbee
+# (C) 2015-2024
+#
+# Initial authors: zaraki673 & pipiche38
+#
+# SPDX-License-Identifier:    GPL-3.0 license
+
+import contextlib
 import hashlib
 import json
 import os
 import platform
 import re
-import sys
 import time
-
+from Modules.tools import how_many_devices
 import distro
 import requests
 
@@ -58,9 +70,7 @@ def populate_custom_dimmensions(self):
         _custom_dimensions[ "dimension4"] = clean_custom_dimension_value( _coordinator_version)
 
     # Network Size
-    total, router, end_devices = get_network_size_items(self.pluginParameters.get("NetworkSize"))
-    if total:
-        _custom_dimensions[ "dimension5"] = clean_custom_dimension_value(total)
+    _custom_dimensions[ "dimension5"] = clean_custom_dimension_value(get_network_size_items(self))
 
     # Certified Db Version
     certified_db_version = self.pluginParameters.get("CertifiedDbVersion")
@@ -148,7 +158,7 @@ def send_matomo_request(self, action_name, custom_variable=None, custom_dimensio
     client_id = get_clientid(self)
     self.log.logging( "Matomo", "Debug", f"send_matomo_request - Clien_id {client_id}")
     if client_id is None:
-        self.log.logging( "Matomo", "Error", "Noting reported as MacAddress is None!")
+        self.log.logging( "Matomo", "Error", "Nothing reported as MacAddress is None!")
         return
 
     # Construct the payload
@@ -260,28 +270,18 @@ def get_uptime_category(start_time):
     return classify_uptime(uptime_seconds)
 
 
-def get_network_size_items(networksize):
-
-    if not networksize:
-        return None, None, None
-
-    # Split the string by " | " to separate each category
-    parts = networksize.split(" | ")
-
-    # Extract individual values using string splitting
-    try:
-        networkTotalsize = int(parts[0].split(":")[1].strip())  # Extract the number after "Total: "
-        networkRoutersize = int(parts[1].split(":")[1].strip())  # Extract the number after "Routers: "
-        networkEndDevicesize = int(parts[2].split(":")[1].strip())  # Extract the number after "End Devices: "
-
-    except IndexError:
-        print("Error: The string format doesn't match the expected pattern.")
-        return None, None, None
+def get_network_size_items(self):
 
-    return classify_nwk_size(networkTotalsize), classify_nwk_size(networkRoutersize), classify_nwk_size(networkEndDevicesize)
+    routers, end_devices = how_many_devices(self)
+    networkTotalsize = routers + end_devices
+
+    return classify_nwk_size(networkTotalsize)
 
 
 def classify_nwk_size(value):
+    if value == 0:
+        return "unknown"
+
     if value < 5:
         return "Micro"
     elif 5 <= value < 10:
@@ -292,8 +292,8 @@ def classify_nwk_size(value):
         return "Large"
     elif 50 <= value < 75:
         return "Very Large"
-    else:
-        return "Xtra Large"
+
+    return "Xtra Large"
 
 
 def get_distribution(self):
@@ -319,7 +319,24 @@ def clean_custom_dimension_value(value: str) -> str:
 
 
 def get_raspberry_pi_model():
+    """ Return PlatformId if it exist """
     if os.path.exists( DEVICE_TREE_CONFIGURATION ):
         with open( DEVICE_TREE_CONFIGURATION , 'r') as f:
             return f.read().strip()
+    return None
+
+
+def get_platform_id():
+    """ Retrieve platform-specific details like hardware """
+
+    pi_model = get_raspberry_pi_model()
+    if pi_model:
+        return pi_model
+
+    # Extract hardware and revision from /proc/cpuinfo
+    with contextlib.suppress(FileNotFoundError):
+        with open("/proc/cpuinfo", "r") as cpuinfo_file:
+            for line in cpuinfo_file:
+                if line.startswith("Hardware"):
+                    return line.split(":")[1].strip()
     return None

Modules/pluginHelpers.py

@@ -210,7 +210,7 @@ def parse_constraints(home_folder):
         "bellows": ""
     }
 
-    constraints_file = Path(home_folder) / "constraints.txt.txt"
+    constraints_file = Path(home_folder) / "constraints.txt"
     with open(constraints_file, 'r') as file:
         for line in file:
             # Remove leading/trailing whitespace and newlines