Replace broken QR login with manual JSESSIONID cookie
Samsung rebuilt account.samsung.com as a JS SPA backed by /iam/oauth2, so the integration's HTML-scraping QR flow can no longer find the signin URL and /accounts/v1/FMM2/signInWithQrCode now 404s. The STF backend (chkLogin.do, getDeviceList.do, ...) is unchanged. Replace the multi-step QR config flow with a single form that asks the user to paste the JSESSIONID cookie copied from a logged-in browser session at smartthingsfind.samsung.com. validate_jsessionid hits chkLogin.do to verify the cookie before accepting it. Also bundle in earlier compatibility fixes for newer HA: store devices in hass.data so the coordinator can look them up by entry_id, and use async_on_update on the device_tracker entity. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
13bd1dabe0
commit
13ff5a534e
7 changed files with 126 additions and 133 deletions
|
|
@ -60,7 +60,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
# fetch data from STF and update the device_tracker and sensor
|
# fetch data from STF and update the device_tracker and sensor
|
||||||
# entities
|
# entities
|
||||||
update_interval = entry.options.get(CONF_UPDATE_INTERVAL, CONF_UPDATE_INTERVAL_DEFAULT)
|
update_interval = entry.options.get(CONF_UPDATE_INTERVAL, CONF_UPDATE_INTERVAL_DEFAULT)
|
||||||
coordinator = SmartThingsFindCoordinator(hass, session, devices, update_interval)
|
coordinator = SmartThingsFindCoordinator(hass, session, entry.entry_id, update_interval)
|
||||||
|
|
||||||
# This is what makes the whole integration slow to load (around 10-15
|
# This is what makes the whole integration slow to load (around 10-15
|
||||||
# seconds for my 15 devices) but it is the right way to do it. Only if
|
# seconds for my 15 devices) but it is the right way to do it. Only if
|
||||||
|
|
@ -92,10 +92,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
class SmartThingsFindCoordinator(DataUpdateCoordinator):
|
class SmartThingsFindCoordinator(DataUpdateCoordinator):
|
||||||
"""Class to manage fetching SmartThings Find data."""
|
"""Class to manage fetching SmartThings Find data."""
|
||||||
|
|
||||||
def __init__(self, hass: HomeAssistant, session: aiohttp.ClientSession, devices, update_interval : int):
|
def __init__(self, hass: HomeAssistant, session: aiohttp.ClientSession, entry_id: str, update_interval : int):
|
||||||
"""Initialize the coordinator."""
|
"""Initialize the coordinator."""
|
||||||
self.session = session
|
self.session = session
|
||||||
self.devices = devices
|
self._entry_id = entry_id
|
||||||
self.hass = hass
|
self.hass = hass
|
||||||
super().__init__(
|
super().__init__(
|
||||||
hass,
|
hass,
|
||||||
|
|
@ -107,11 +107,12 @@ class SmartThingsFindCoordinator(DataUpdateCoordinator):
|
||||||
async def _async_update_data(self):
|
async def _async_update_data(self):
|
||||||
"""Fetch data from SmartThings Find."""
|
"""Fetch data from SmartThings Find."""
|
||||||
try:
|
try:
|
||||||
|
devices = self.hass.data[DOMAIN][self._entry_id]["devices"]
|
||||||
tags = {}
|
tags = {}
|
||||||
_LOGGER.debug(f"Updating locations...")
|
_LOGGER.debug(f"Updating locations...")
|
||||||
for device in self.devices:
|
for device in devices:
|
||||||
dev_data = device['data']
|
dev_data = device['data']
|
||||||
tag_data = await get_device_location(self.hass, self.session, dev_data, self.config_entry.entry_id)
|
tag_data = await get_device_location(self.hass, self.session, dev_data, self._entry_id)
|
||||||
tags[dev_data['dvceID']] = tag_data
|
tags[dev_data['dvceID']] = tag_data
|
||||||
_LOGGER.debug(f"Fetched {len(tags)} locations")
|
_LOGGER.debug(f"Fetched {len(tags)} locations")
|
||||||
return tags
|
return tags
|
||||||
|
|
|
||||||
|
|
@ -1,11 +1,11 @@
|
||||||
from typing import Any, Dict
|
from typing import Any
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.core import callback
|
from homeassistant.core import callback
|
||||||
from homeassistant.config_entries import (
|
from homeassistant.config_entries import (
|
||||||
ConfigEntry,
|
ConfigEntry,
|
||||||
ConfigFlowResult,
|
ConfigFlowResult,
|
||||||
OptionsFlowWithConfigEntry
|
OptionsFlowWithConfigEntry,
|
||||||
)
|
)
|
||||||
from .const import (
|
from .const import (
|
||||||
DOMAIN,
|
DOMAIN,
|
||||||
|
|
@ -15,130 +15,65 @@ from .const import (
|
||||||
CONF_ACTIVE_MODE_SMARTTAGS,
|
CONF_ACTIVE_MODE_SMARTTAGS,
|
||||||
CONF_ACTIVE_MODE_SMARTTAGS_DEFAULT,
|
CONF_ACTIVE_MODE_SMARTTAGS_DEFAULT,
|
||||||
CONF_ACTIVE_MODE_OTHERS,
|
CONF_ACTIVE_MODE_OTHERS,
|
||||||
CONF_ACTIVE_MODE_OTHERS_DEFAULT
|
CONF_ACTIVE_MODE_OTHERS_DEFAULT,
|
||||||
)
|
)
|
||||||
from .utils import do_login_stage_one, do_login_stage_two, gen_qr_code_base64
|
from .utils import validate_jsessionid
|
||||||
import asyncio
|
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class SmartThingsFindConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
class SmartThingsFindConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
"""Handle a config flow for SmartThings Find."""
|
"""Handle a config flow for SmartThings Find.
|
||||||
|
|
||||||
|
Samsung's account.samsung.com login was rebuilt as a JS SPA (IAM/OAuth2
|
||||||
|
with bot protection), so the original QR-code login flow no longer works.
|
||||||
|
Until that is reverse-engineered we fall back to letting the user paste
|
||||||
|
the JSESSIONID cookie they obtain from a normal browser session at
|
||||||
|
https://smartthingsfind.samsung.com/.
|
||||||
|
"""
|
||||||
|
|
||||||
VERSION = 1
|
VERSION = 1
|
||||||
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
|
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
|
||||||
|
|
||||||
reauth_entry: ConfigEntry | None = None
|
reauth_entry: ConfigEntry | None = None
|
||||||
|
|
||||||
task_stage_one: asyncio.Task | None = None
|
async def _async_handle_jsessionid(self, jsessionid: str) -> ConfigFlowResult:
|
||||||
task_stage_two: asyncio.Task | None = None
|
data = {CONF_JSESSIONID: jsessionid}
|
||||||
|
|
||||||
qr_url = None
|
|
||||||
session = None
|
|
||||||
|
|
||||||
jsessionid = None
|
|
||||||
|
|
||||||
|
|
||||||
error = None
|
|
||||||
|
|
||||||
async def do_stage_one(self):
|
|
||||||
_LOGGER.debug("Running login stage 1")
|
|
||||||
try:
|
|
||||||
stage_one_res = await do_login_stage_one(self.hass)
|
|
||||||
if not stage_one_res is None:
|
|
||||||
self.session, self.qr_url = stage_one_res
|
|
||||||
else:
|
|
||||||
self.error = "Login stage 1 failed. Check logs for details."
|
|
||||||
_LOGGER.warn("Login stage 1 failed")
|
|
||||||
_LOGGER.debug("Login stage 1 done")
|
|
||||||
except Exception as e:
|
|
||||||
self.error = "Login stage 1 failed. Check logs for details."
|
|
||||||
_LOGGER.error(f"Exception in stage 1: {e}", exc_info=True)
|
|
||||||
|
|
||||||
async def do_stage_two(self):
|
|
||||||
_LOGGER.debug("Running login stage 2")
|
|
||||||
try:
|
|
||||||
stage_two_res = await do_login_stage_two(self.session)
|
|
||||||
if not stage_two_res is None:
|
|
||||||
self.jsessionid = stage_two_res
|
|
||||||
_LOGGER.info("Login successful")
|
|
||||||
else:
|
|
||||||
self.error = "Login stage 2 failed. Check logs for details."
|
|
||||||
_LOGGER.warning("Login stage 2 failed")
|
|
||||||
_LOGGER.debug("Login stage 2 done")
|
|
||||||
except Exception as e:
|
|
||||||
self.error = "Login stage 2 failed. Check logs for details."
|
|
||||||
_LOGGER.error(f"Exception in stage 2: {e}", exc_info=True)
|
|
||||||
|
|
||||||
# First step: Get QR Code login URL
|
|
||||||
async def async_step_user(self, user_input=None):
|
|
||||||
_LOGGER.debug("Entering login stage 1")
|
|
||||||
if not self.task_stage_one:
|
|
||||||
self.task_stage_one = self.hass.async_create_task(self.do_stage_one())
|
|
||||||
if not self.task_stage_one.done():
|
|
||||||
return self.async_show_progress(
|
|
||||||
progress_action="task_stage_one",
|
|
||||||
progress_task=self.task_stage_one
|
|
||||||
)
|
|
||||||
# At this point stage 1 is completed
|
|
||||||
if self.error:
|
|
||||||
# An error occurred, cancel the flow by moving to the finish-
|
|
||||||
# form which shows the error
|
|
||||||
return self.async_show_progress_done(next_step_id="finish")
|
|
||||||
# No error -> Proceed to stage 2
|
|
||||||
return self.async_show_progress_done(next_step_id="auth_stage_two")
|
|
||||||
|
|
||||||
# Second step: Wait until QR scanned an log in
|
|
||||||
async def async_step_auth_stage_two(self, user_input=None):
|
|
||||||
if not self.task_stage_two:
|
|
||||||
self.task_stage_two = self.hass.async_create_task(self.do_stage_two())
|
|
||||||
if not self.task_stage_two.done():
|
|
||||||
return self.async_show_progress(
|
|
||||||
progress_action="task_stage_two",
|
|
||||||
progress_task=self.task_stage_two,
|
|
||||||
description_placeholders={
|
|
||||||
"qr_code": gen_qr_code_base64(self.qr_url),
|
|
||||||
"url": self.qr_url,
|
|
||||||
"code": self.qr_url.split('/')[-1],
|
|
||||||
}
|
|
||||||
)
|
|
||||||
return self.async_show_progress_done(next_step_id="finish")
|
|
||||||
|
|
||||||
async def async_step_finish(self, user_input=None):
|
|
||||||
if self.error:
|
|
||||||
return self.async_show_form(step_id="finish", errors={'base': self.error})
|
|
||||||
|
|
||||||
data={CONF_JSESSIONID: self.jsessionid}
|
|
||||||
|
|
||||||
if self.reauth_entry:
|
if self.reauth_entry:
|
||||||
# Finish step was called by reauth-flow. Do not create a new entry,
|
return self.async_update_reload_and_abort(self.reauth_entry, data=data)
|
||||||
# instead update the existing entry
|
|
||||||
return self.async_update_reload_and_abort(
|
|
||||||
self.reauth_entry,
|
|
||||||
data=data
|
|
||||||
)
|
|
||||||
|
|
||||||
return self.async_create_entry(title="SmartThings Find", data=data)
|
return self.async_create_entry(title="SmartThings Find", data=data)
|
||||||
|
|
||||||
|
async def async_step_user(self, user_input=None):
|
||||||
|
errors: dict[str, str] = {}
|
||||||
|
if user_input is not None:
|
||||||
|
jsessionid = (user_input.get(CONF_JSESSIONID) or "").strip()
|
||||||
|
if not jsessionid:
|
||||||
|
errors["base"] = "empty_cookie"
|
||||||
|
elif await validate_jsessionid(self.hass, jsessionid):
|
||||||
|
return await self._async_handle_jsessionid(jsessionid)
|
||||||
|
else:
|
||||||
|
errors["base"] = "invalid_auth"
|
||||||
|
|
||||||
|
return self.async_show_form(
|
||||||
|
step_id="user",
|
||||||
|
data_schema=vol.Schema({vol.Required(CONF_JSESSIONID): str}),
|
||||||
|
errors=errors,
|
||||||
|
)
|
||||||
|
|
||||||
async def async_step_reauth(self, user_input=None):
|
async def async_step_reauth(self, user_input=None):
|
||||||
self.reauth_entry = self.hass.config_entries.async_get_entry(
|
self.reauth_entry = self.hass.config_entries.async_get_entry(
|
||||||
self.context["entry_id"]
|
self.context["entry_id"]
|
||||||
)
|
)
|
||||||
return await self.async_step_reauth_confirm()
|
|
||||||
|
|
||||||
async def async_step_reauth_confirm(self, user_input=None):
|
|
||||||
if user_input is None:
|
|
||||||
return self.async_show_form(
|
|
||||||
step_id="user",
|
|
||||||
data_schema=vol.Schema({}),
|
|
||||||
)
|
|
||||||
return await self.async_step_user()
|
return await self.async_step_user()
|
||||||
|
|
||||||
async def async_step_reconfigure(self, user_input: dict[str, Any] | None = None):
|
async def async_step_reconfigure(self, user_input: dict[str, Any] | None = None):
|
||||||
return await self.async_step_reauth_confirm(self)
|
self.reauth_entry = self.hass.config_entries.async_get_entry(
|
||||||
|
self.context["entry_id"]
|
||||||
|
)
|
||||||
|
return await self.async_step_user()
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@callback
|
@callback
|
||||||
def async_get_options_flow(
|
def async_get_options_flow(
|
||||||
|
|
@ -146,8 +81,8 @@ class SmartThingsFindConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
) -> config_entries.OptionsFlow:
|
) -> config_entries.OptionsFlow:
|
||||||
"""Create the options flow."""
|
"""Create the options flow."""
|
||||||
return SmartThingsFindOptionsFlowHandler(config_entry)
|
return SmartThingsFindOptionsFlowHandler(config_entry)
|
||||||
|
|
||||||
|
|
||||||
class SmartThingsFindOptionsFlowHandler(OptionsFlowWithConfigEntry):
|
class SmartThingsFindOptionsFlowHandler(OptionsFlowWithConfigEntry):
|
||||||
"""Handle an options flow."""
|
"""Handle an options flow."""
|
||||||
|
|
||||||
|
|
@ -186,4 +121,4 @@ class SmartThingsFindOptionsFlowHandler(OptionsFlowWithConfigEntry):
|
||||||
): bool,
|
): bool,
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
return self.async_show_form(step_id="init", data_schema=data_schema)
|
return self.async_show_form(step_id="init", data_schema=data_schema)
|
||||||
|
|
|
||||||
|
|
@ -43,7 +43,7 @@ class SmartThingsDeviceTracker(DeviceTrackerEntity):
|
||||||
|
|
||||||
if 'icons' in device['data'] and 'coloredIcon' in device['data']['icons']:
|
if 'icons' in device['data'] and 'coloredIcon' in device['data']['icons']:
|
||||||
self._attr_entity_picture = device['data']['icons']['coloredIcon']
|
self._attr_entity_picture = device['data']['icons']['coloredIcon']
|
||||||
self.async_update = coordinator.async_add_listener(self.async_write_ha_state)
|
self.async_on_update(coordinator.async_add_listener(self.async_write_ha_state))
|
||||||
|
|
||||||
def async_write_ha_state(self):
|
def async_write_ha_state(self):
|
||||||
if not self.enabled:
|
if not self.enabled:
|
||||||
|
|
|
||||||
|
|
@ -2,13 +2,13 @@
|
||||||
"domain": "smartthings_find",
|
"domain": "smartthings_find",
|
||||||
"name": "SmartThings Find",
|
"name": "SmartThings Find",
|
||||||
"after_dependencies": ["http"],
|
"after_dependencies": ["http"],
|
||||||
"version": "0.2.1",
|
"version": "0.3.0",
|
||||||
"documentation": "https://github.com/Vedeneb/HA-SmartThings-Find",
|
"documentation": "https://git.jeena.net/jeena/HA-SmartThings-Find",
|
||||||
"issue_tracker": "https://github.com/Vedeneb/HA-SmartThings-Find/issues",
|
"issue_tracker": "https://git.jeena.net/jeena/HA-SmartThings-Find/issues",
|
||||||
"integration_type": "hub",
|
"integration_type": "hub",
|
||||||
"dependencies": [],
|
"dependencies": [],
|
||||||
"codeowners": ["@Vedeneb"],
|
"codeowners": ["@jeena"],
|
||||||
"requirements": ["requests", "qrcode"],
|
"requirements": ["requests", "qrcode[pil]", "pillow", "pytz"],
|
||||||
"iot_class": "cloud_polling",
|
"iot_class": "cloud_polling",
|
||||||
"config_flow": true
|
"config_flow": true
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -3,9 +3,18 @@
|
||||||
"abort": {
|
"abort": {
|
||||||
"reauth_successful": "Erfolgreich authentifiziert."
|
"reauth_successful": "Erfolgreich authentifiziert."
|
||||||
},
|
},
|
||||||
"progress": {
|
"error": {
|
||||||
"task_stage_one": "Vorbereitung läuft, bitte warten...",
|
"invalid_auth": "Das Cookie wurde von SmartThings Find abgelehnt. Stelle sicher, dass du nur den Wert des JSESSIONID-Cookies (nicht den gesamten Cookie-Header) aus einer angemeldeten Sitzung kopiert hast.",
|
||||||
"task_stage_two": "Bitte scanne den QR Code mit deinem Galaxy-Gerät, um dich anzumelden. Alternativ kannst du dich direkt im Browser anmelden, indum du [hier]({url}) klickst. \n\n\n\nCode kann nicht gescannt werden? Gehe auf [signin.samsung.com/key](https://signin.samsung.com/key/) und gib den folgenden Code ein:\n\n## ```{code}```\n"
|
"empty_cookie": "Bitte einen JSESSIONID-Wert einfügen."
|
||||||
|
},
|
||||||
|
"step": {
|
||||||
|
"user": {
|
||||||
|
"title": "JSESSIONID-Cookie einfügen",
|
||||||
|
"description": "Samsungs QR-Code-Login funktioniert nicht mehr (account.samsung.com wurde umgebaut). Bis das ersetzt ist, musst du das Session-Cookie manuell besorgen:\n\n1. Öffne [smartthingsfind.samsung.com](https://smartthingsfind.samsung.com/) in einem normalen Browser und melde dich an.\n2. Öffne DevTools → Anwendung (Chrome) bzw. Speicher (Firefox) → Cookies → smartthingsfind.samsung.com.\n3. Kopiere den Wert des **JSESSIONID**-Cookies und füge ihn unten ein.\n\nDas Cookie bleibt meist mehrere Wochen gültig. Läuft es ab, fordert Home Assistant eine erneute Authentifizierung an – dann diese Schritte einfach wiederholen.",
|
||||||
|
"data": {
|
||||||
|
"jsessionid": "JSESSIONID"
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"options": {
|
"options": {
|
||||||
|
|
|
||||||
|
|
@ -3,9 +3,18 @@
|
||||||
"abort": {
|
"abort": {
|
||||||
"reauth_successful": "Reauthentication successful."
|
"reauth_successful": "Reauthentication successful."
|
||||||
},
|
},
|
||||||
"progress": {
|
"error": {
|
||||||
"task_stage_one": "Preparing, please wait...",
|
"invalid_auth": "Cookie was rejected by SmartThings Find. Make sure you copied the JSESSIONID value (not the whole cookie header) from a tab where you are logged in.",
|
||||||
"task_stage_two": "To login please scan the following QR Code with your Galaxy device. You can also login using this browser by clicking [here]({url}). \n\n\n\nUnable to scan the code? Go to [signin.samsung.com/key](https://signin.samsung.com/key/) and enter the following code:\n\n## ```{code}```\n"
|
"empty_cookie": "Please paste a JSESSIONID value."
|
||||||
|
},
|
||||||
|
"step": {
|
||||||
|
"user": {
|
||||||
|
"title": "Paste JSESSIONID cookie",
|
||||||
|
"description": "Samsung's QR code login flow no longer works (account.samsung.com was rebuilt). Until it is replaced you have to obtain the session cookie manually:\n\n1. Open [smartthingsfind.samsung.com](https://smartthingsfind.samsung.com/) in a regular browser and log in.\n2. Open DevTools → Application (Chrome) or Storage (Firefox) → Cookies → smartthingsfind.samsung.com.\n3. Copy the value of the **JSESSIONID** cookie and paste it below.\n\nThe cookie usually stays valid for several weeks. When it expires Home Assistant will trigger a reauth and you repeat these steps.",
|
||||||
|
"data": {
|
||||||
|
"jsessionid": "JSESSIONID"
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"options": {
|
"options": {
|
||||||
|
|
@ -19,4 +28,4 @@
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -211,6 +211,28 @@ async def do_login_stage_two(session: aiohttp.ClientSession) -> str:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
async def validate_jsessionid(hass: HomeAssistant, jsessionid: str) -> bool:
|
||||||
|
"""Check whether a JSESSIONID cookie is currently valid for SmartThings Find.
|
||||||
|
|
||||||
|
Performs a single GET against the CSRF endpoint with the cookie applied;
|
||||||
|
a valid session returns 200 with a `_csrf` response header.
|
||||||
|
"""
|
||||||
|
session = async_get_clientsession(hass)
|
||||||
|
session.cookie_jar.update_cookies({"JSESSIONID": jsessionid})
|
||||||
|
try:
|
||||||
|
async with session.get(URL_GET_CSRF) as response:
|
||||||
|
if response.status == 200 and response.headers.get("_csrf"):
|
||||||
|
return True
|
||||||
|
body = (await response.text())[:200]
|
||||||
|
_LOGGER.warning(
|
||||||
|
f"JSESSIONID validation failed: status={response.status}, body='{body}'"
|
||||||
|
)
|
||||||
|
return False
|
||||||
|
except Exception as e:
|
||||||
|
_LOGGER.error(f"JSESSIONID validation error: {e}", exc_info=True)
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
async def fetch_csrf(hass: HomeAssistant, session: aiohttp.ClientSession, entry_id: str):
|
async def fetch_csrf(hass: HomeAssistant, session: aiohttp.ClientSession, entry_id: str):
|
||||||
"""
|
"""
|
||||||
Retrieves the _csrf-Token which needs to be sent with each following request.
|
Retrieves the _csrf-Token which needs to be sent with each following request.
|
||||||
|
|
@ -377,8 +399,13 @@ async def get_device_location(hass: HomeAssistant, session: aiohttp.ClientSessio
|
||||||
utcDate = parse_stf_date(
|
utcDate = parse_stf_date(
|
||||||
op['extra']['gpsUtcDt'])
|
op['extra']['gpsUtcDt'])
|
||||||
else:
|
else:
|
||||||
_LOGGER.error(
|
_LOGGER.warning(
|
||||||
f"[{dev_name}] No UTC date found for operation '{op['oprnType']}', this should not happen! OP: {json.dumps(op)}")
|
f"[{dev_name}] No UTC date found for operation '{op['oprnType']}'. OP: {json.dumps(op)}")
|
||||||
|
continue
|
||||||
|
|
||||||
|
if utcDate is None:
|
||||||
|
_LOGGER.warning(
|
||||||
|
f"[{dev_name}] Could not parse UTC date for operation '{op['oprnType']}'")
|
||||||
continue
|
continue
|
||||||
|
|
||||||
if used_loc['gps_date'] and used_loc['gps_date'] >= utcDate:
|
if used_loc['gps_date'] and used_loc['gps_date'] >= utcDate:
|
||||||
|
|
@ -419,6 +446,10 @@ async def get_device_location(hass: HomeAssistant, session: aiohttp.ClientSessio
|
||||||
continue
|
continue
|
||||||
else:
|
else:
|
||||||
utcDate = parse_stf_date(loc['gpsUtcDt'])
|
utcDate = parse_stf_date(loc['gpsUtcDt'])
|
||||||
|
if utcDate is None:
|
||||||
|
_LOGGER.warning(
|
||||||
|
f"[{dev_name}] Could not parse UTC date for encrypted location ('{op['oprnType']}')")
|
||||||
|
continue
|
||||||
if used_loc['gps_date'] and used_loc['gps_date'] >= utcDate:
|
if used_loc['gps_date'] and used_loc['gps_date'] >= utcDate:
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
f"[{dev_name}] Ignoring location older than the previous ({op['oprnType']})")
|
f"[{dev_name}] Ignoring location older than the previous ({op['oprnType']})")
|
||||||
|
|
@ -519,17 +550,18 @@ def get_sub_location(ops: list, subDeviceName: str) -> tuple:
|
||||||
for op in ops:
|
for op in ops:
|
||||||
if subDeviceName in op.get('encLocation', {}):
|
if subDeviceName in op.get('encLocation', {}):
|
||||||
loc = op['encLocation'][subDeviceName]
|
loc = op['encLocation'][subDeviceName]
|
||||||
|
gps_date = parse_stf_date(loc.get('gpsUtcDt', ''))
|
||||||
sub_loc = {
|
sub_loc = {
|
||||||
"latitude": float(loc['latitude']),
|
"latitude": float(loc['latitude']) if loc.get('latitude') is not None else None,
|
||||||
"longitude": float(loc['longitude']),
|
"longitude": float(loc['longitude']) if loc.get('longitude') is not None else None,
|
||||||
"gps_accuracy": calc_gps_accuracy(loc.get('horizontalUncertainty'), loc.get('verticalUncertainty')),
|
"gps_accuracy": calc_gps_accuracy(loc.get('horizontalUncertainty'), loc.get('verticalUncertainty')),
|
||||||
"gps_date": parse_stf_date(loc['gpsUtcDt'])
|
"gps_date": gps_date
|
||||||
}
|
}
|
||||||
return op, sub_loc
|
return op, sub_loc
|
||||||
return {}, {}
|
return {}, {}
|
||||||
|
|
||||||
|
|
||||||
def parse_stf_date(datestr: str) -> datetime:
|
def parse_stf_date(datestr: str) -> datetime | None:
|
||||||
"""
|
"""
|
||||||
Parses a date string in the format "%Y%m%d%H%M%S" to a datetime object.
|
Parses a date string in the format "%Y%m%d%H%M%S" to a datetime object.
|
||||||
This is the format, the SmartThings Find API uses.
|
This is the format, the SmartThings Find API uses.
|
||||||
|
|
@ -538,9 +570,16 @@ def parse_stf_date(datestr: str) -> datetime:
|
||||||
datestr (str): The date string in the format "%Y%m%d%H%M%S".
|
datestr (str): The date string in the format "%Y%m%d%H%M%S".
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
datetime: A datetime object representing the input date string.
|
datetime: A datetime object representing the input date string,
|
||||||
|
or None if parsing fails.
|
||||||
"""
|
"""
|
||||||
return datetime.strptime(datestr, "%Y%m%d%H%M%S").replace(tzinfo=pytz.UTC)
|
if not datestr:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
return datetime.strptime(datestr, "%Y%m%d%H%M%S").replace(tzinfo=pytz.UTC)
|
||||||
|
except ValueError:
|
||||||
|
_LOGGER.warning(f"Failed to parse date string: '{datestr}'")
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def get_battery_level(dev_name: str, ops: list) -> int:
|
def get_battery_level(dev_name: str, ops: list) -> int:
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue