@@ -1541,6 +1541,154 @@ async def on_off_departure_times(data, vehicle, turn_on:bool) -> bool:
15411541 else :
15421542 return await vehicle .departure_times_disable ()
15431543
1544+ # FIRMWARE / OTA UPDATE SENSORS
1545+ # the actual IVSU installation progress lives under states.deployment - it walks through
1546+ # artifact_retrieval_in_progress -> installation_queued -> deploying -> success (or failure), and
1547+ # while "deploying" the message payload carries an ETA + 12V battery telemetry (Ford aborts the
1548+ # install if the 12V battery is too weak). states.configurationUpdate is a separate, much less
1549+ # frequent config-sync record that we fall back to if no deployment has been seen yet
1550+ def get_firmware_update_status_state (data , prev_state = None ):
1551+ states = FordpassDataHandler .get_states (data )
1552+ to_state = states .get ("deployment" , {}).get ("value" , {}).get ("toState" )
1553+ if to_state is None :
1554+ to_state = states .get ("configurationUpdate" , {}).get ("value" , {}).get ("toState" )
1555+ return to_state if to_state is not None else UNSUPPORTED
1556+
1557+ def get_firmware_update_status_attrs (data , units ):
1558+ states = FordpassDataHandler .get_states (data )
1559+ deployment = states .get ("deployment" , {})
1560+ attrs = {}
1561+
1562+ if deployment :
1563+ value = deployment .get ("value" , {})
1564+ attrs ["fromState" ] = value .get ("fromState" )
1565+ attrs ["timestamp" ] = deployment .get ("timestamp" )
1566+ attrs ["commandId" ] = deployment .get ("commandId" )
1567+
1568+ try :
1569+ telemetry = json .loads (deployment .get ("message" , "" ))
1570+ except (ValueError , TypeError ):
1571+ telemetry = None
1572+ if isinstance (telemetry , dict ):
1573+ if "estimatedTimeToActivate" in telemetry :
1574+ attrs ["estimatedSecondsToActivate" ] = telemetry ["estimatedTimeToActivate" ]
1575+ if "currentLvBatterySoc" in telemetry :
1576+ attrs ["lvBatterySocDuringUpdate" ] = telemetry ["currentLvBatterySoc" ]
1577+
1578+ config_update = states .get ("configurationUpdate" , {})
1579+ if config_update :
1580+ attrs ["lastConfigurationUpdateTimestamp" ] = config_update .get ("timestamp" )
1581+
1582+ return attrs
1583+
1584+ # the weekly OTA activation window is only available via states.commands.getASUSettingsCommand -
1585+ # this gets populated whenever ANY client (FordPass app or us) asked Ford for it, we don't have to
1586+ # request it ourselves
1587+ def get_ota_schedule_state (data , prev_state = None ):
1588+ oem_data = (FordpassDataHandler .get_states (data )
1589+ .get ("commands" , {})
1590+ .get ("getASUSettingsCommand" , {})
1591+ .get ("value" , {})
1592+ .get ("oemData" , {}))
1593+ day_schedule_strs = oem_data .get ("OTAActivationDaySchedule" , {}).get ("stringArrayValue" , [])
1594+ if not day_schedule_strs :
1595+ return None
1596+
1597+ now = datetime .now ()
1598+ next_activation = None
1599+ for entry_str in day_schedule_strs :
1600+ try :
1601+ entry = json .loads (entry_str )
1602+ except (ValueError , TypeError ):
1603+ continue
1604+
1605+ day_of_week = entry .get ("activationDayOfWeek" , "" ).upper ()
1606+ if day_of_week not in DAYS_MAP :
1607+ continue
1608+
1609+ for a_time in entry .get ("activationScheduleTime" , []):
1610+ target_weekday = DAYS_MAP [day_of_week ]
1611+ days_until = (target_weekday - now .weekday () + 7 ) % 7
1612+ candidate = (now + timedelta (days = days_until )).replace (
1613+ hour = a_time .get ("hour" , 0 ),
1614+ minute = a_time .get ("minute" , 0 ),
1615+ second = a_time .get ("second" , 0 ),
1616+ microsecond = 0 ,
1617+ )
1618+ if candidate <= now :
1619+ candidate += timedelta (days = 7 )
1620+ if next_activation is None or candidate < next_activation :
1621+ next_activation = candidate
1622+
1623+ if next_activation is None :
1624+ return None
1625+
1626+ local_tz = datetime .now ().astimezone ().tzinfo
1627+ return next_activation .replace (tzinfo = local_tz )
1628+
1629+ def get_ota_schedule_attrs (data , units ):
1630+ oem_data = (FordpassDataHandler .get_states (data )
1631+ .get ("commands" , {})
1632+ .get ("getASUSettingsCommand" , {})
1633+ .get ("value" , {})
1634+ .get ("oemData" , {}))
1635+ if not oem_data :
1636+ return {}
1637+
1638+ weekly_schedule = {}
1639+ for entry_str in oem_data .get ("OTAActivationDaySchedule" , {}).get ("stringArrayValue" , []):
1640+ try :
1641+ entry = json .loads (entry_str )
1642+ except (ValueError , TypeError ):
1643+ continue
1644+ day = entry .get ("activationDayOfWeek" , "" ).lower ()
1645+ times = entry .get ("activationScheduleTime" , [])
1646+ if day and times :
1647+ weekly_schedule [day ] = ", " .join (
1648+ f"{ a_time .get ('hour' , 0 ):02d} { a_time .get ('minute' , 0 ):02d} for a_time in times
1649+ )
1650+
1651+ return {
1652+ "autoSoftwareUpdateEnabled" : oem_data .get ("ASUState" , {}).get ("stringValue" ),
1653+ "scheduleType" : oem_data .get ("activationScheduleSetting" , {}).get ("stringValue" ),
1654+ "weeklySchedule" : weekly_schedule ,
1655+ }
1656+
1657+ # the last successfully installed ECU/firmware package, sourced from the (non-custom) top-level
1658+ # events.configurationUpdateEvent
1659+ def get_last_firmware_update_state (data , prev_state = None ):
1660+ config_update_event = FordpassDataHandler .get_events (data ).get ("configurationUpdateEvent" , {})
1661+ update_time = config_update_event .get ("updateTime" )
1662+ if update_time is None :
1663+ return None
1664+ try :
1665+ return dt .as_local (dt .parse_datetime (update_time ))
1666+ except BaseException as ex :
1667+ _LOGGER .debug (f"get_last_firmware_update_state(): could not parse datetime: '{ update_time } { ex } )
1668+ return None
1669+
1670+ def get_last_firmware_update_attrs (data , units ):
1671+ config_update_event = FordpassDataHandler .get_events (data ).get ("configurationUpdateEvent" , {})
1672+ if not config_update_event :
1673+ return {}
1674+
1675+ oem_data = config_update_event .get ("oemData" , {})
1676+ updated_ecus = []
1677+ for entry_str in oem_data .get ("ecu_configuration" , {}).get ("stringArrayValue" , []):
1678+ try :
1679+ ecu = json .loads (entry_str )
1680+ except (ValueError , TypeError ):
1681+ continue
1682+ updated_ecus .append ({
1683+ "ecuId" : ecu .get ("ECUId" ),
1684+ "partNumber" : ecu .get ("partIIPartNumber" ),
1685+ })
1686+
1687+ return {
1688+ "firmwareVersion" : oem_data .get ("ftcp_version" , {}).get ("stringValue" ),
1689+ "updatedEcus" : updated_ecus ,
1690+ }
1691+
15441692 # DEPARTURE_SCHEDULE SENSOR & SERVICE stuff
15451693 def get_departure_schedules_state (data , prev_state = None ):
15461694 ev_attrs = FordpassDataHandler .get_elveh_attrs (data , units = None )
0 commit comments