Skip to content

Commit

Permalink
In zigpy mode, the OTA is much faster. Looks like we do 7 blocks/second
Browse files Browse the repository at this point in the history
  • Loading branch information
pipiche38 committed Nov 30, 2024
1 parent 6ee2e45 commit e4c375a
Showing 1 changed file with 19 additions and 11 deletions.
30 changes: 19 additions & 11 deletions Classes/OTA.py
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,7 @@ class OTAManagement(object):

def __init__(
self,
zigbee_communitation,
zigbee_communication,
PluginConf,
DeviceConf,
adminWidgets,
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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)


Expand Down

0 comments on commit e4c375a

Please sign in to comment.