diff --git a/.gitignore b/.gitignore index 592f206..b728a17 100644 --- a/.gitignore +++ b/.gitignore @@ -74,3 +74,6 @@ venv.bak/ # Generated response files response-*.json + +.copilot* +external diff --git a/pyproject.toml b/pyproject.toml index c24b9d5..fcdda6e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -77,7 +77,8 @@ exclude = [ [tool.ruff.per-file-ignores] "docs/source/conf.py" = ["D100"] -"pysamrthashtag/api/authentication.py" = ["D102", "D107"] +"pysmarthashtag/api/authentication.py" = ["D102", "D107"] +"pysmarthashtag/vehicle/battery.py" = ["C901"] # Complex data parsing function [tool.ruff.mccabe] max-complexity = 25 diff --git a/pysmarthashtag/account.py b/pysmarthashtag/account.py index de347fd..7fa056c 100644 --- a/pysmarthashtag/account.py +++ b/pysmarthashtag/account.py @@ -7,10 +7,10 @@ from typing import Optional from pysmarthashtag.api import utils -from pysmarthashtag.api.authentication import SmartAuthentication +from pysmarthashtag.api.authentication import SmartAuthentication, SmartAuthenticationINTL from pysmarthashtag.api.client import SmartClient, SmartClientConfiguration from pysmarthashtag.api.log_sanitizer import sanitize_log_data -from pysmarthashtag.const import API_CARS_URL, API_SELECT_CAR_URL, EndpointUrls +from pysmarthashtag.const import API_CARS_URL, API_SELECT_CAR_URL, EndpointUrls, SmartRegion from pysmarthashtag.models import SmartAuthError, SmartHumanCarConnectionError, SmartTokenRefreshNecessary from pysmarthashtag.vehicle.vehicle import SmartVehicle @@ -38,21 +38,72 @@ class SmartAccount: endpoint_urls: Optional[EndpointUrls] = None """Optional. Custom endpoint URLs for international API support.""" + region: Optional[SmartRegion] = None + """Optional. Region preset (EU or INTL). If set, overrides endpoint_urls.""" + vehicles: dict[str, SmartVehicle] = field(default_factory=dict, init=False) """Vehicles associated with the account.""" def __post_init__(self, password, log_responses): """Initialize the account.""" - # Ensure endpoint_urls is set - if self.endpoint_urls is None: + # Import get_endpoint_urls_for_region here to avoid circular imports + from pysmarthashtag.const import get_endpoint_urls_for_region + + # If region is specified, use it to get endpoint URLs + if self.region is not None: + self.endpoint_urls = get_endpoint_urls_for_region(self.region) + elif self.endpoint_urls is None: self.endpoint_urls = EndpointUrls() + # Choose authentication class based on region + if self.region == SmartRegion.INTL: + _LOGGER.info("Using INTL (International) authentication for region: %s", self.region) + auth = SmartAuthenticationINTL(self.username, password, endpoint_urls=self.endpoint_urls) + else: + _LOGGER.info("Using EU authentication (default)") + auth = SmartAuthentication(self.username, password, endpoint_urls=self.endpoint_urls) + if self.config is None: self.config = SmartClientConfiguration( - SmartAuthentication(self.username, password, endpoint_urls=self.endpoint_urls), + auth, log_responses=log_responses, ) + def _is_intl_region(self) -> bool: + """Check if using INTL (International) region.""" + return self.region == SmartRegion.INTL + + def _generate_api_headers(self, params: dict, method: str, url: str, body=None) -> dict[str, str]: + """Generate API headers based on region (EU or INTL). + + This method automatically selects the correct header generator + based on whether the account is configured for INTL region. + """ + auth = self.config.authentication + + if self._is_intl_region(): + # INTL region uses different headers and signing + client_id = getattr(auth, "api_client_id", None) + return utils.generate_intl_header( + device_id=auth.device_id, + access_token=auth.api_access_token, + params=params, + method=method, + url=url, + body=body, + client_id=client_id, + ) + else: + # EU region uses default headers + return utils.generate_default_header( + device_id=auth.device_id, + access_token=auth.api_access_token, + params=params, + method=method, + url=url, + body=body, + ) + async def _ensure_ssl_context(self) -> None: """Ensure SSL context is created asynchronously. @@ -90,9 +141,7 @@ async def _init_vehicles(self) -> None: # we do not know what type of car we have in our list so we fall back to the old API URL self.endpoint_urls.get_api_base_url() + API_CARS_URL + "?" + utils.join_url_params(params), headers={ - **utils.generate_default_header( - client.config.authentication.device_id, - client.config.authentication.api_access_token, + **self._generate_api_headers( params=params, method="GET", url=API_CARS_URL, @@ -152,9 +201,7 @@ async def select_active_vehicle(self, vin) -> None: r_car_info = await client.post( self.vehicles[vin].base_url + API_SELECT_CAR_URL, headers={ - **utils.generate_default_header( - client.config.authentication.device_id, - client.config.authentication.api_access_token, + **self._generate_api_headers( params={}, method="POST", url=API_SELECT_CAR_URL, @@ -176,28 +223,41 @@ async def select_active_vehicle(self, vin) -> None: async def get_vehicle_information(self, vin) -> str: """Get information about a vehicle.""" _LOGGER.debug("Getting information for vehicle") - params = { - "latest": True, - "target": "basic%2Cmore", - "userId": self.config.authentication.api_user_id, - } + + # INTL uses different API path and parameters than EU + if self._is_intl_region(): + # INTL: No /api/v1/ prefix, latest=False to get all data including climate/doors/pollution + params = { + "latest": False, + "target": "basic,more", + "userId": self.config.authentication.api_user_id, + } + url_path = "/remote-control/vehicle/status/" + vin + # INTL uses api.ecloudeu.com without /api/v1/ prefix + base_url = self.endpoint_urls.get_api_base_url() + else: + # EU: Uses /api/v1/ prefix in the base_url setting + params = { + "latest": True, + "target": "basic,more", + "userId": self.config.authentication.api_user_id, + } + url_path = "/remote-control/vehicle/status/" + vin + base_url = self.vehicles[vin].base_url + data = {} async with SmartClient(self.config) as client: for retry in range(3): try: + # Build URL with URL-encoded params for GET request + url_params = {k: str(v).replace(",", "%2C") for k, v in params.items()} r_car_info = await client.get( - self.vehicles[vin].base_url - + "/remote-control/vehicle/status/" - + vin - + "?" - + utils.join_url_params(params), + base_url + url_path + "?" + utils.join_url_params(url_params), headers={ - **utils.generate_default_header( - client.config.authentication.device_id, - client.config.authentication.api_access_token, + **self._generate_api_headers( params=params, method="GET", - url="/remote-control/vehicle/status/" + vin, + url=url_path, ) }, ) @@ -233,9 +293,7 @@ async def get_vehicle_soc(self, vin) -> str: + "?" + utils.join_url_params(params), headers={ - **utils.generate_default_header( - client.config.authentication.device_id, - client.config.authentication.api_access_token, + **self._generate_api_headers( params=params, method="GET", url="/remote-control/vehicle/status/soc/" + vin, diff --git a/pysmarthashtag/api/authentication.py b/pysmarthashtag/api/authentication.py index 4124d7b..88472d1 100644 --- a/pysmarthashtag/api/authentication.py +++ b/pysmarthashtag/api/authentication.py @@ -19,6 +19,11 @@ from pysmarthashtag.const import ( API_SESION_URL, HTTPX_TIMEOUT, + INTL_APP_ID, + INTL_CA_KEY, + INTL_LOGIN_URL, + INTL_OAUTH_URL, + INTL_OPERATOR_CODE, EndpointUrls, ) from pysmarthashtag.models import SmartAPIError @@ -390,6 +395,293 @@ async def async_auth_flow(self, request: Request) -> AsyncGenerator[Request, Res raise +class SmartAuthenticationINTL(SmartAuthentication): + """Authentication handler for Smart International (INTL) API. + + This class handles authentication for the Hello Smart International app, + which is used in Australia, Singapore, Israel, and other international markets. + + The INTL auth flow is different from EU: + 1. POST login with email/password to sg-app-api.smart.com -> get accessToken, idToken + 2. GET oauth20/authorize with accessToken -> get authCode + 3. POST session/secure with authCode to apiv2.ecloudeu.com -> get vehicle API tokens + + Note: The sg-app-api.smart.com endpoints do not require X-Ca-Signature validation, + which simplifies the authentication process significantly. + """ + + def __init__( + self, + username: str, + password: str, + access_token: Optional[str] = None, + expires_at: Optional[datetime.datetime] = None, + refresh_token: Optional[str] = None, + ssl_context: Optional[ssl.SSLContext] = None, + endpoint_urls: Optional[EndpointUrls] = None, + ): + super().__init__( + username=username, + password=password, + access_token=access_token, + expires_at=expires_at, + refresh_token=refresh_token, + ssl_context=ssl_context, + endpoint_urls=endpoint_urls, + ) + # INTL-specific tokens + self.intl_access_token: Optional[str] = None + self.intl_refresh_token: Optional[str] = None + self.intl_id_token: Optional[str] = None + self.intl_user_id: Optional[str] = None + self.session_id: str = secrets.token_hex(16).upper() + self.device_identifier: str = secrets.token_hex(16) + self.client_id: str = "2232193363e44527" # INTL OAuth client ID + self.api_client_id: Optional[str] = None # Client ID from session response + _LOGGER.debug("SmartAuthenticationINTL initialized for INTL region") + + def _generate_intl_headers(self, auth_token: str = "") -> dict[str, str]: + """Generate headers for INTL API requests (sg-app-api.smart.com). + + The sg-app-api.smart.com endpoints do not validate X-Ca-Signature, + so we can use empty or minimal signatures. + """ + import time + import uuid + + timestamp = str(int(time.time() * 1000)) + nonce = str(uuid.uuid4()).upper() + + headers = { + "Host": "sg-app-api.smart.com", + "User-Agent": "GlobalSmart/1.0.7 (iPhone; iOS 18.6.1; Scale/3.00)", + "X-Ca-Key": INTL_CA_KEY, + "X-Ca-Timestamp": timestamp, + "X-Ca-Nonce": nonce, + "X-Ca-Signature-Method": "HmacSHA256", + "X-Ca-Signature": "", # Not validated by server + "X-Ca-Signature-Headers": ( + "Accept-Language,User-Agent,X-Ca-Nonce,X-Ca-Timestamp," + "Xs-App-Ver,Xs-Auth-Token,Xs-Client-Id,Xs-Di,Xs-Os,Xs-Session-Id,Xs-Ui" + ), + "Xs-Os": "iOS", + "Xs-App-Ver": "1.0.7", + "Xs-Session-Id": self.session_id, + "Xs-Di": self.device_identifier, + "Xs-Ui": secrets.token_hex(16), + "Xs-Auth-Token": auth_token, + "Xs-Client-Id": self.client_id, + "Accept": "*/*", + "Accept-Language": "en-AU;q=1", + "Content-Type": "application/json", + "Cache-Control": "no-cache", + } + + return headers + + def _generate_vehicle_api_headers(self, body: Optional[str] = None) -> dict[str, str]: + """Generate headers for vehicle API (ecloudeu) requests. + + These headers are used for the session/secure endpoint on apiv2.ecloudeu.com. + The x-signature is calculated using the INTL signing secret. + """ + import time + import uuid + + timestamp = str(int(time.time() * 1000)) + nonce = str(uuid.uuid4()).upper() + + headers = { + "x-app-id": INTL_APP_ID, + "x-operator-code": INTL_OPERATOR_CODE, + "x-agent-type": "iOS", + "x-agent-version": "18.6.1", + "x-device-type": "mobile", + "x-device-identifier": self.device_identifier, + "x-device-manufacture": "Apple", + "x-device-brand": "Apple", + "x-device-model": "iPhone", + "x-env-type": "production", + "x-api-signature-version": "1.0", + "x-api-signature-nonce": nonce, + "x-timestamp": timestamp, + "platform": "NON-CMA", + "accept": "application/json;responseformat=3", + "accept-language": "en_US", + "content-type": "application/json", + "user-agent": "GlobalSmart/1.0.7 (iPhone; iOS 18.6.1; Scale/3.00)", + } + + # Calculate signature using INTL secret + if body is not None: + headers["x-signature"] = utils._create_sign( + nonce=nonce, + params={"identity_type": "smart-israel"}, + timestamp=timestamp, + method="POST", + url=API_SESION_URL, + body=body, + use_intl=True, + ) + + return headers + + async def _login(self): + """Login to Smart INTL web services. + + INTL login flow (uses sg-app-api.smart.com which doesn't require signatures): + 1. POST to login endpoint with email/password -> accessToken, idToken + 2. GET oauth20/authorize with accessToken -> authCode + 3. POST session/secure to apiv2.ecloudeu.com with authCode -> vehicle API tokens + """ + ssl_ctx = await self.get_ssl_context() + async with SmartLoginClient(ssl_context=ssl_ctx) as client: + _LOGGER.info("INTL: Acquiring access token via Hello Smart International API") + + # Step 1: Login with email/password (no signature required!) + _LOGGER.debug("INTL: Step 1 - Logging in with credentials") + login_data = json.dumps( + { + "email": self.username, + "password": self.password, + } + ) + + r_login = await client.post( + INTL_LOGIN_URL, + content=login_data, + headers=self._generate_intl_headers(), + ) + + try: + login_result = r_login.json() + _LOGGER.debug("INTL login response code: %s", login_result.get("code")) + + if login_result.get("code") != "200": + error_msg = login_result.get("message", "Unknown error") + _LOGGER.error("INTL login failed: %s", error_msg) + raise SmartAPIError(f"INTL login failed: {error_msg}") + + result = login_result.get("result", {}) + self.intl_access_token = result.get("accessToken") + self.intl_refresh_token = result.get("refreshToken") + self.intl_id_token = result.get("idToken") + self.intl_user_id = result.get("userId") + + # Validate that required tokens are present + missing_tokens = [] + if not self.intl_access_token: + missing_tokens.append("accessToken") + if not self.intl_id_token: + missing_tokens.append("idToken") + if not self.intl_user_id: + missing_tokens.append("userId") + + if missing_tokens: + error_msg = f"INTL login response missing required tokens: {', '.join(missing_tokens)}" + _LOGGER.error(error_msg) + raise SmartAPIError(error_msg) + + expires_in = result.get("expiresIn", 86400) + + expires_at = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=expires_in) + + _LOGGER.info("INTL: Login successful for user %s", self.intl_user_id) + + except (KeyError, ValueError, TypeError) as e: + raise SmartAPIError(f"Could not parse INTL login response: {e}") from e + + # Step 2: Exchange accessToken for authCode via OAuth + # Note: OAuth uses just accessToken in URL param, and idToken in Xs-Auth-Token header + _LOGGER.debug("INTL: Step 2 - Exchanging accessToken for authCode") + + oauth_url = f"{INTL_OAUTH_URL}?accessToken={self.intl_access_token}" + + oauth_headers = self._generate_intl_headers(self.intl_id_token) + + r_oauth = await client.get( + oauth_url, + headers=oauth_headers, + ) + + try: + oauth_result = r_oauth.json() + _LOGGER.debug("INTL OAuth response received.") + + if oauth_result.get("code") != "200": + error_msg = oauth_result.get("message", "Unknown error") + sanitized_error_msg = sanitize_log_data(error_msg) + _LOGGER.error("INTL OAuth failed: %s", sanitized_error_msg) + raise SmartAPIError(f"INTL OAuth failed: {error_msg}") + + # Result is directly the authCode string (e.g., "CODE-xxx") + auth_code = oauth_result.get("result") + if isinstance(auth_code, dict): + auth_code = auth_code.get("authCode", auth_code) + + _LOGGER.debug("INTL: Got authCode.") + + except (KeyError, ValueError, TypeError) as e: + raise SmartAPIError(f"Could not parse INTL OAuth response: {e}") from e + + # Step 3: Exchange authCode for vehicle API session + _LOGGER.debug("INTL: Step 3 - Exchanging authCode for vehicle API session") + + session_url = f"{self.endpoint_urls.get_api_base_url_v2()}{API_SESION_URL}?identity_type=smart-israel" + session_data = json.dumps({"authCode": auth_code}, separators=(",", ":")) + + session_headers = self._generate_vehicle_api_headers(body=session_data) + + r_session = await client.post( + session_url, + content=session_data, + headers=session_headers, + ) + + try: + session_result = r_session.json() + _LOGGER.debug("INTL Session response: %s", sanitize_log_data(session_result)) + + if session_result.get("code") != 1000: + error_msg = session_result.get("message", "Unknown error") + sanitized_code = sanitize_log_data(session_result.get("code")) + _LOGGER.error("INTL session failed: %s (code: %s)", sanitize_log_data(error_msg), sanitized_code) + raise SmartAPIError(f"INTL session failed: {error_msg}") + + data = session_result.get("data", {}) + api_access_token = data.get("accessToken") + api_refresh_token = data.get("refreshToken") + api_user_id = data.get("userId") + api_client_id = data.get("clientId") + + # Store client_id on the auth object for use in API requests + self.api_client_id = api_client_id + + _LOGGER.info("INTL: Successfully authenticated, user ID: %s", sanitize_log_data(api_user_id)) + + except (KeyError, ValueError, TypeError) as e: + raise SmartAPIError(f"Could not parse INTL session response: {e}") from e + + # Return OAuth tokens for standard access, and session API tokens for vehicle API + return { + "access_token": self.intl_access_token, + "refresh_token": self.intl_refresh_token, + "api_access_token": api_access_token, + "api_refresh_token": api_refresh_token, + "api_user_id": api_user_id, + "expires_at": expires_at, + } + + async def _refresh_access_token(self): + """Refresh the INTL access token. + + For INTL, we simply re-login as the refresh flow is complex + and the login endpoint doesn't require signatures. + """ + _LOGGER.debug("INTL: Refresh requested, performing full re-login") + return {} # Return empty to trigger full login + + def get_retry_wait_time(response: httpx.Response) -> int: """Get the wait time to wait twice as long before retrying.""" try: diff --git a/pysmarthashtag/api/log_sanitizer.py b/pysmarthashtag/api/log_sanitizer.py index 7fad5da..ae2cdf9 100644 --- a/pysmarthashtag/api/log_sanitizer.py +++ b/pysmarthashtag/api/log_sanitizer.py @@ -126,8 +126,41 @@ def _sanitize_list(data: list, depth: int = 0, max_depth: int = 10) -> list: return result +def _looks_like_secret(s: str) -> bool: + """Check if a string looks like a secret (high entropy, no whitespace, base64/hex-like). + + Args: + ---- + s: String to check + + Returns: + ------- + True if the string appears to be a secret/token + + """ + # If it has whitespace, it's likely a human-readable message, not a secret + if any(c.isspace() for c in s): + return False + + # Long strings with only alphanumeric characters and common encoding chars + # (likely base64, hex, or URL-safe tokens) + alphanumeric_chars = sum(1 for c in s if c.isalnum()) + special_chars = sum(1 for c in s if c in "-_=+/") + total_chars = len(s) + + # If mostly alphanumeric with some special chars, and no whitespace, looks like a secret + if total_chars > 20 and (alphanumeric_chars + special_chars) / total_chars > 0.95: + return True + + # Check for common secret patterns: long hex strings, base64-like strings + if len(s) >= 32 and all(c in "0123456789abcdefABCDEF" for c in s): + return True # Hex string + + return False + + def _sanitize_string(data: str) -> str: - """Sanitize a string by masking VINs and tokens. + """Sanitize a string by masking VINs, tokens, and potentially sensitive content. Args: ---- @@ -142,6 +175,18 @@ def _sanitize_string(data: str) -> str: result = VIN_PATTERN.sub(lambda m: _mask_value(m.group()), data) # Mask Bearer tokens result = TOKEN_PATTERN.sub(r"\1***", result) + + # As a defensive fallback, avoid logging very long or opaque strings in full. + # This helps when upstream services accidentally include secrets in generic + # "message" fields that do not match our specific patterns. + # Only truncate if it looks like a secret and is long enough + max_visible_length = 512 + if len(result) > max_visible_length and _looks_like_secret(result): + # Preserve only a short prefix/suffix to keep logs useful while hiding content. + prefix = result[:40] + suffix = result[-10:] + return f"{prefix}***{suffix}" + return result @@ -166,7 +211,15 @@ def sanitize_log_data(data: Any) -> Any: return _sanitize_list(data) elif isinstance(data, str): return _sanitize_string(data) - return data + + # For any other type (including numbers, custom objects, etc.), avoid + # returning the raw value to prevent accidental leakage of sensitive data. + # Allowlist safe primitives while keeping generic sanitization for other types. + if isinstance(data, (int, float, bool)): + return str(data) + if data is None: + return "None" + return f"" def get_data_summary(data: dict, include_keys: Union[list, None] = None) -> str: diff --git a/pysmarthashtag/api/utils.py b/pysmarthashtag/api/utils.py index 1e32318..cc65064 100644 --- a/pysmarthashtag/api/utils.py +++ b/pysmarthashtag/api/utils.py @@ -4,19 +4,63 @@ import logging import secrets import time +from typing import Optional +from urllib.parse import quote _LOGGER = logging.getLogger(__name__) +# EU signing secret (base64 encoded) +EU_SECRET = base64.b64decode("NzRlNzQ2OWFmZjUwNDJiYmJlZDdiYmIxYjM2YzE1ZTk=") + +# INTL (International) signing secret - used for Australia, Singapore, Israel, etc. +INTL_SECRET = b"30fb4a7b7fab4e2e8b52120d0087efdd" + def join_url_params(args: dict) -> str: """Join params for adding to URL.""" return "&".join([f"{key}={value}" for key, value in args.items()]) -def _create_sign(nonce: str, params: dict, timestamp: str, method: str, url: str, body=None) -> str: - """Create a signature for the request.""" +def _create_sign( + nonce: str, + params: dict, + timestamp: str, + method: str, + url: str, + body: Optional[str] = None, + use_intl: bool = False, + url_encode_params: bool = False, +) -> str: + """Create a signature for the request. + + Args: + ---- + nonce: Random nonce for the request + params: URL parameters as a dict + timestamp: Timestamp in milliseconds + method: HTTP method (GET, POST, etc.) + url: API URL path (without domain) + body: Request body for POST requests + use_intl: If True, use INTL secret instead of EU secret + url_encode_params: If True, URL-encode special chars in params (required for INTL GET requests) + + Returns: + ------- + Base64-encoded HMAC-SHA1 signature. + + """ md5sum = base64.b64encode(hashlib.md5(body.encode()).digest()).decode() if body else "1B2M2Y8AsgTpgAmY7PhCfg==" - url_params = join_url_params(params) + + # URL encode params if needed (INTL GET requests require this for special chars like commas) + if url_encode_params and params: + encoded_params = {} + for k, v in params.items(): + # URL encode special characters (commas, spaces, etc.) + encoded_params[k] = quote(str(v), safe="") + url_params = join_url_params(encoded_params) + else: + url_params = join_url_params(params) if params else "" + payload = f"""application/json;responseformat=3 x-api-signature-nonce:{nonce} x-api-signature-version:1.0 @@ -26,16 +70,19 @@ def _create_sign(nonce: str, params: dict, timestamp: str, method: str, url: str {timestamp} {method} {url}""" - _LOGGER.debug("Creating signature for request") - secret = base64.b64decode("NzRlNzQ2OWFmZjUwNDJiYmJlZDdiYmIxYjM2YzE1ZTk=") - payload = payload.encode("utf-8") - hashed = hmac.new(secret, payload, hashlib.sha1).digest() + _LOGGER.debug("Creating signature for request (INTL=%s)", use_intl) + + # Choose secret based on region + secret = INTL_SECRET if use_intl else EU_SECRET + + payload_bytes = payload.encode("utf-8") + hashed = hmac.new(secret, payload_bytes, hashlib.sha1).digest() signature = base64.b64encode(hashed).decode() return signature def generate_default_header( - device_id: str, access_token: str, params: dict, method: str, url: str, body=None + device_id: str, access_token: str, params: dict, method: str, url: str, body: Optional[str] = None ) -> dict[str, str]: """Generate a header for HTTP requests to the server.""" timestamp = create_correct_timestamp() @@ -69,6 +116,82 @@ def generate_default_header( return header +def generate_intl_header( + device_id: str, + access_token: str, + params: dict, + method: str, + url: str, + body: Optional[str] = None, + client_id: Optional[str] = None, +) -> dict[str, str]: + """Generate a header for HTTP requests to the INTL (International) API. + + This is used for vehicles registered in Australia, Singapore, Israel, + and other international markets using the Hello Smart International app. + + Args: + ---- + device_id: Unique device identifier + access_token: API access token from session/secure + params: URL parameters as a dict + method: HTTP method (GET, POST, etc.) + url: API URL path (without domain) + body: Request body for POST requests + client_id: Client ID from session response + + Returns: + ------- + Dict of headers for the request. + + """ + import uuid + + timestamp = create_correct_timestamp() + nonce = str(uuid.uuid4()).upper() + + # INTL requires URL-encoded params in signature for GET requests with special chars + url_encode_params = method.upper() == "GET" + sign = _create_sign(nonce, params, timestamp, method, url, body, use_intl=True, url_encode_params=url_encode_params) + + # Get vehicle series from params if available (passed as _vehicle_series) + vehicle_series = params.pop("_vehicle_series", None) if params else None + if not vehicle_series: + raise ValueError("vehicle_series is required for INTL API requests. Pass it in params as '_vehicle_series'.") + + header = { + "x-app-id": "SMARTAPP-ISRAEL", + "accept": "application/json;responseformat=3", + "x-agent-type": "iOS", + "x-device-type": "mobile", + "x-operator-code": "SMART-ISRAEL", + "x-device-identifier": device_id, + "x-env-type": "production", + "accept-language": "en_US", + "x-api-signature-version": "1.0", + "x-api-signature-nonce": nonce, + "x-device-manufacture": "Apple", + "x-device-brand": "Apple", + "x-device-model": "iPhone", + "x-agent-version": "18.6.1", + "content-type": "application/json", + "user-agent": "GlobalSmart/1.0.7 (iPhone; iOS 18.6.1; Scale/3.00)", + "x-signature": sign, + "x-timestamp": timestamp, + "platform": "NON-CMA", + "x-vehicle-series": vehicle_series, + } + + if access_token: + header["authorization"] = access_token + + if client_id: + header["x-client-id"] = client_id + + _LOGGER.debug("Constructed INTL request header for %s %s", method, url) + return header + + def create_correct_timestamp() -> str: """Create a correct timestamp for the request.""" return str(int(time.time() * 1000)) diff --git a/pysmarthashtag/const.py b/pysmarthashtag/const.py index 4ae6da1..4385efa 100644 --- a/pysmarthashtag/const.py +++ b/pysmarthashtag/const.py @@ -1,6 +1,7 @@ """URLs for different services and error code mapping.""" from dataclasses import dataclass +from enum import Enum from typing import Optional API_KEY = "3_L94eyQ-wvJhWm7Afp1oBhfTGXZArUfSHHW9p9Pncg513hZELXsxCfMWHrF8f5P5a" @@ -14,11 +15,78 @@ API_SELECT_CAR_URL = "/device-platform/user/session/update" API_TELEMATICS_URL = "/remote-control/vehicle/telematics/" +# INTL (International) API specific constants +# Used for Hello Smart International app (Australia, Singapore, Israel, etc.) +INTL_API_BASE = "https://sg-app-api.smart.com" +INTL_LOGIN_URL = f"{INTL_API_BASE}/iam/service/api/v1/login" +INTL_OAUTH_URL = f"{INTL_API_BASE}/iam/service/api/v1/oauth20/authorize" +INTL_REFRESH_URL = f"{INTL_API_BASE}/iam/service/api/v1/refresh/" +INTL_USER_INFO_URL = f"{INTL_API_BASE}/uc/api/user/toc/base/info" +INTL_CA_KEY = "204587190" # X-Ca-Key header value for INTL API Gateway +INTL_APP_ID = "SMARTAPP-ISRAEL" # x-app-id for vehicle API +INTL_OPERATOR_CODE = "SMART-ISRAEL" # x-operator-code for vehicle API +# Note: INTL signing secret is stored in api/utils.py + OTA_SERVER_URL = "https://ota.srv.smart.com/" HTTPX_TIMEOUT = 30.0 +class SmartRegion(str, Enum): + """Region presets for Smart API endpoints. + + Use these region presets to easily configure the API for different geographic regions. + - EU: European region (default) - for users with Hello Smart EU app + - INTL: International region (Asia-Pacific) - for users with Hello Smart International app + (Australia, Singapore, and other international markets) + """ + + EU = "eu" + INTL = "intl" + + +def get_endpoint_urls_for_region(region: SmartRegion) -> "EndpointUrls": + """Get pre-configured EndpointUrls for a specific region. + + Args: + ---- + region: The region to get endpoint URLs for. + + Returns: + ------- + EndpointUrls configured for the specified region. + + Example: + ------- + >>> from pysmarthashtag.const import SmartRegion, get_endpoint_urls_for_region + >>> from pysmarthashtag.account import SmartAccount + >>> + >>> # For Australian/International users + >>> endpoint_urls = get_endpoint_urls_for_region(SmartRegion.INTL) + >>> account = SmartAccount("user@example.com", "password", endpoint_urls=endpoint_urls) + >>> + >>> # For European users (default) + >>> endpoint_urls = get_endpoint_urls_for_region(SmartRegion.EU) + >>> account = SmartAccount("user@example.com", "password", endpoint_urls=endpoint_urls) + + """ + if region == SmartRegion.EU: + # European region - uses default endpoints + return EndpointUrls() + elif region == SmartRegion.INTL: + # International region (Asia-Pacific) - for Hello Smart International app + # Used in Australia, Singapore, Israel, and other international markets + # Note: INTL uses different auth system (sg-app-api.smart.com) but same data endpoints as EU + return EndpointUrls( + server_url=INTL_LOGIN_URL, + auth_url=INTL_OAUTH_URL, + login_url=INTL_LOGIN_URL, + # INTL uses same data endpoints as EU (leave as None to use defaults) + ) + else: + raise ValueError(f"Unknown region: {region}") + + @dataclass class EndpointUrls: """Configuration for API endpoint URLs. diff --git a/pysmarthashtag/tests/test_endpoint_urls.py b/pysmarthashtag/tests/test_endpoint_urls.py index c42a94a..5084471 100644 --- a/pysmarthashtag/tests/test_endpoint_urls.py +++ b/pysmarthashtag/tests/test_endpoint_urls.py @@ -13,6 +13,8 @@ OTA_SERVER_URL, SERVER_URL, EndpointUrls, + SmartRegion, + get_endpoint_urls_for_region, ) from pysmarthashtag.tests import TEST_PASSWORD, TEST_USERNAME from pysmarthashtag.tests.conftest import prepare_account_with_vehicles @@ -170,3 +172,64 @@ async def test_account_with_explicit_default_urls(smart_fixture: respx.Router): await account.get_vehicles() assert account is not None assert len(account.vehicles) == 2 + + +class TestSmartRegion: + """Test SmartRegion enum and get_endpoint_urls_for_region function.""" + + def test_region_eu_returns_default_endpoints(self): + """Test that EU region returns default (European) endpoints.""" + urls = get_endpoint_urls_for_region(SmartRegion.EU) + assert urls.get_api_base_url() == API_BASE_URL + assert urls.get_api_base_url_v2() == "https://apiv2.ecloudeu.com" + assert urls.get_server_url() == SERVER_URL + assert urls.get_login_url() == LOGIN_URL + assert urls.get_auth_url() == AUTH_URL + assert urls.get_ota_server_url() == OTA_SERVER_URL + + def test_region_intl_returns_asia_pacific_endpoints(self): + """Test that INTL region returns EU endpoints (same cloud infrastructure).""" + urls = get_endpoint_urls_for_region(SmartRegion.INTL) + # International region uses EU cloud endpoints (shared infrastructure) + assert urls.get_api_base_url() == "https://api.ecloudeu.com" + assert urls.get_api_base_url_v2() == "https://apiv2.ecloudeu.com" + # INTL uses different auth URLs (sg-app-api.smart.com) + assert urls.get_login_url() == "https://sg-app-api.smart.com/iam/service/api/v1/login" + assert urls.get_auth_url() == "https://sg-app-api.smart.com/iam/service/api/v1/oauth20/authorize" + assert urls.get_server_url() == "https://sg-app-api.smart.com/iam/service/api/v1/login" + + def test_region_enum_values(self): + """Test that SmartRegion enum has expected values.""" + assert SmartRegion.EU.value == "eu" + assert SmartRegion.INTL.value == "intl" + + def test_account_with_eu_region(self): + """Test creating SmartAccount with EU region preset.""" + urls = get_endpoint_urls_for_region(SmartRegion.EU) + account = SmartAccount( + username="test@example.com", + password="testpass", + endpoint_urls=urls, + ) + assert account.endpoint_urls.get_api_base_url() == API_BASE_URL + + def test_account_with_intl_region(self): + """Test creating SmartAccount with INTL region preset.""" + urls = get_endpoint_urls_for_region(SmartRegion.INTL) + account = SmartAccount( + username="test@example.com", + password="testpass", + endpoint_urls=urls, + ) + assert account.endpoint_urls.get_api_base_url() == "https://api.ecloudeu.com" + assert account.endpoint_urls.get_api_base_url_v2() == "https://apiv2.ecloudeu.com" + + def test_authentication_with_intl_region(self): + """Test SmartAuthentication with INTL region preset.""" + urls = get_endpoint_urls_for_region(SmartRegion.INTL) + auth = SmartAuthentication( + username="test@example.com", + password="testpass", + endpoint_urls=urls, + ) + assert auth.endpoint_urls.get_api_base_url() == "https://api.ecloudeu.com" diff --git a/pysmarthashtag/tests/test_missing_fields.py b/pysmarthashtag/tests/test_missing_fields.py index 0f61372..0c62ae4 100644 --- a/pysmarthashtag/tests/test_missing_fields.py +++ b/pysmarthashtag/tests/test_missing_fields.py @@ -2,7 +2,10 @@ import logging +import pytest + from pysmarthashtag.models import get_field_as_type +from pysmarthashtag.vehicle.basic_status import BasicStatus from pysmarthashtag.vehicle.battery import Battery from pysmarthashtag.vehicle.climate import Climate from pysmarthashtag.vehicle.maintenance import Maintenance @@ -10,6 +13,7 @@ from pysmarthashtag.vehicle.running import Running from pysmarthashtag.vehicle.safety import Safety from pysmarthashtag.vehicle.tires import Tires +from pysmarthashtag.vehicle.trailer import Trailer class TestGetFieldAsType: @@ -270,3 +274,134 @@ def test_dc_charging_negative_value(self, caplog): assert battery.charging_voltage is not None # Should use the first element in DcChargingVoltLevels assert battery.charging_voltage.value == DcChargingVoltLevels[0] + + +class TestTrailerExceptionHandling: + """Test that Trailer exception handling works correctly after fixing finally block.""" + + def test_trailer_keyerror_is_caught_and_logged(self, caplog): + """Test that KeyError exceptions are caught and logged, not propagated.""" + # Missing 'additionalVehicleStatus' will cause KeyError on line 52 + vehicle_data = { + "vehicleStatus": { + # Missing 'additionalVehicleStatus' + "updateTime": "1706028240000", + } + } + with caplog.at_level(logging.INFO): + # Should not raise KeyError, but should log it + result = Trailer.from_vehicle_data(vehicle_data) + + # KeyError should be caught and logged + assert "Trailer status info not available" in caplog.text + # Should return None since no data was parsed + assert result is None + + def test_trailer_non_keyerror_propagates(self): + """Test that non-KeyError exceptions (e.g., ValueError) propagate correctly.""" + # Invalid updateTime that will cause ValueError in int() conversion + vehicle_data = { + "vehicleStatus": { + "additionalVehicleStatus": { + "trailerStatus": { + "trailerTurningLampSts": "0", + } + }, + "updateTime": "not_a_number", # Will cause ValueError in int() + } + } + + # ValueError should propagate, not be swallowed + with pytest.raises(ValueError): + Trailer.from_vehicle_data(vehicle_data) + + def test_trailer_with_valid_data_returns_object(self): + """Test that valid trailer data parses correctly.""" + vehicle_data = { + "vehicleStatus": { + "additionalVehicleStatus": { + "trailerStatus": { + "trailerTurningLampSts": "0", + "trailerFogLampSts": "1", + "trailerBreakLampSts": "0", + "trailerReversingLampSts": "0", + "trailerPosLampSts": "1", + } + }, + "updateTime": "1706028240000", + } + } + + trailer = Trailer.from_vehicle_data(vehicle_data) + assert trailer is not None + assert trailer.turning_lamp_status == 0 + assert trailer.fog_lamp_status == 1 + + +class TestBasicStatusExceptionHandling: + """Test that BasicStatus exception handling works correctly after fixing finally block.""" + + def test_basic_status_from_vehicle_data_returns_none_when_empty(self, caplog): + """Test that BasicStatus.from_vehicle_data returns None when no data can be parsed.""" + # With empty vehicleStatus, BasicStatus uses .get() so it won't raise KeyError + # Instead, it should return None since no meaningful data was parsed + vehicle_data = { + "vehicleStatus": {} # Minimal data that won't trigger parsing + } + with caplog.at_level(logging.INFO): + # Should not raise KeyError, should return None + result = BasicStatus.from_vehicle_data(vehicle_data) + + # Should return None since no meaningful data was parsed + assert result is None + + def test_basic_status_non_keyerror_propagates(self): + """Test that non-KeyError exceptions (e.g., ValueError) propagate correctly.""" + # Invalid updateTime that will cause ValueError in int() conversion + vehicle_data = { + "vehicleStatus": { + "basicVehicleStatus": { + "speed": "50.0", + }, + "updateTime": "invalid_timestamp", # Will cause ValueError in int() + } + } + + # ValueError should propagate, not be swallowed + with pytest.raises(ValueError): + BasicStatus.from_vehicle_data(vehicle_data) + + def test_basic_status_with_invalid_notification_time_propagates(self): + """Test that ValueError from datetime.fromtimestamp propagates.""" + # Very large timestamp value that will cause OSError/ValueError + vehicle_data = { + "vehicleStatus": { + "notification": { + "time": 999999999999999999999, # Absurdly large timestamp + }, + "updateTime": "1706028240000", + } + } + + # Should propagate the exception from datetime.fromtimestamp + with pytest.raises((ValueError, OSError, OverflowError)): + BasicStatus.from_vehicle_data(vehicle_data) + + def test_basic_status_with_valid_data_returns_object(self): + """Test that valid basic status data parses correctly.""" + vehicle_data = { + "vehicleStatus": { + "basicVehicleStatus": { + "speed": "50.0", + "direction": "N", + "engineStatus": "engine_off", + }, + "updateTime": "1706028240000", + } + } + + basic_status = BasicStatus.from_vehicle_data(vehicle_data) + assert basic_status is not None + assert basic_status.speed is not None + assert basic_status.speed.value == 50.0 + assert basic_status.direction == "N" diff --git a/pysmarthashtag/vehicle/basic_status.py b/pysmarthashtag/vehicle/basic_status.py new file mode 100644 index 0000000..2dc37f2 --- /dev/null +++ b/pysmarthashtag/vehicle/basic_status.py @@ -0,0 +1,151 @@ +"""Basic vehicle status models for pysmarthashtag.""" + +import logging +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Optional + +from pysmarthashtag.models import ValueWithUnit, VehicleDataBase, get_field_as_type + +_LOGGER = logging.getLogger(__name__) + + +@dataclass +class BasicStatus(VehicleDataBase): + """Provides an accessible version of the vehicle's basic status data. + + This includes data from basicVehicleStatus, notification, eg (eGuard), + parkTime, and configuration sections of the API response. + """ + + # Basic vehicle status + speed: Optional[ValueWithUnit] = None + """Current vehicle speed.""" + + speed_validity: Optional[bool] = None + """Whether the speed value is valid.""" + + direction: Optional[str] = None + """Vehicle heading direction.""" + + engine_status: Optional[str] = None + """Engine status (e.g., 'engine_off', 'engine_on').""" + + car_mode: Optional[int] = None + """Car mode. 0=normal.""" + + usage_mode: Optional[int] = None + """Usage mode. 0=normal.""" + + car_locator_upload_enabled: Optional[bool] = None + """Whether car locator status upload is enabled.""" + + # Notification (emergency call) status + emergency_call_status: Optional[int] = None + """Emergency call notification status. 0=no active call.""" + + notification_reason: Optional[int] = None + """Reason code for notification.""" + + notification_time: Optional[datetime] = None + """Timestamp of the notification.""" + + notification_parameters: Optional[str] = None + """JSON string of notification parameters.""" + + # eGuard status + eguard_running: Optional[bool] = None + """Whether eGuard is running.""" + + eguard_blocked_status: Optional[int] = None + """eGuard blocked status. 0=not blocked.""" + + panic_status: Optional[bool] = None + """Whether panic mode is active.""" + + # Park time + park_time: Optional[datetime] = None + """Timestamp when the vehicle was last parked.""" + + # Configuration + propulsion_type: Optional[int] = None + """Propulsion type. 4=electric.""" + + fuel_type: Optional[int] = None + """Fuel type. 4=electric.""" + + @classmethod + def from_vehicle_data(cls, vehicle_data: dict): + """Create a new instance based on data from API.""" + parsed = cls._parse_vehicle_data(vehicle_data) or {} + if len(parsed) > 0: + return cls(**parsed) + return None + + @classmethod + def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: + """Parse the basic status data based on the vehicle data dict.""" + _LOGGER.debug("Parsing basic status data") + if "vehicleStatus" not in vehicle_data: + return None + retval: dict[str, Any] = {} + try: + vehicle_status = vehicle_data["vehicleStatus"] + + # Parse basicVehicleStatus + basic_status = vehicle_status.get("basicVehicleStatus", {}) + if basic_status: + speed = get_field_as_type(basic_status, "speed", float) + if speed is not None: + retval["speed"] = ValueWithUnit(speed, "km/h") + retval["speed_validity"] = get_field_as_type(basic_status, "speedValidity", bool) + retval["direction"] = get_field_as_type(basic_status, "direction", str) + retval["engine_status"] = get_field_as_type(basic_status, "engineStatus", str) + retval["car_mode"] = get_field_as_type(basic_status, "carMode", int) + retval["usage_mode"] = get_field_as_type(basic_status, "usageMode", int) + + # Position info within basicVehicleStatus + position = basic_status.get("position", {}) + if position: + car_locator = get_field_as_type(position, "carLocatorStatUploadEn", bool) + retval["car_locator_upload_enabled"] = car_locator + + # Parse notification + notification = vehicle_status.get("notification", {}) + if notification: + retval["emergency_call_status"] = get_field_as_type(notification, "notifForEmgyCallStatus", int) + retval["notification_reason"] = get_field_as_type(notification, "reason", int) + notif_time = get_field_as_type(notification, "time", int) + if notif_time is not None: + retval["notification_time"] = datetime.fromtimestamp(notif_time / 1000.0) + retval["notification_parameters"] = get_field_as_type(notification, "parameters", str) + + # Parse eGuard (eg) + eg = vehicle_status.get("eg", {}) + if eg: + retval["eguard_running"] = get_field_as_type(eg, "enableRunning", bool) + retval["panic_status"] = get_field_as_type(eg, "panicStatus", bool) + blocked = eg.get("blocked", {}) + if blocked: + retval["eguard_blocked_status"] = get_field_as_type(blocked, "status", int) + + # Parse parkTime + park_time = vehicle_status.get("parkTime", {}) + if park_time: + park_status = get_field_as_type(park_time, "status", int) + if park_status is not None: + retval["park_time"] = datetime.fromtimestamp(park_status / 1000) + + # Parse configuration + config = vehicle_status.get("configuration", {}) + if config: + retval["propulsion_type"] = get_field_as_type(config, "propulsionType", int) + retval["fuel_type"] = get_field_as_type(config, "fuelType", int) + + # Timestamp + if "updateTime" in vehicle_status: + retval["timestamp"] = datetime.fromtimestamp(int(vehicle_status["updateTime"]) / 1000) + + except KeyError as e: + _LOGGER.info(f"Basic status info not available: {e}") + return retval diff --git a/pysmarthashtag/vehicle/battery.py b/pysmarthashtag/vehicle/battery.py index 7ba7daf..2e42b9a 100644 --- a/pysmarthashtag/vehicle/battery.py +++ b/pysmarthashtag/vehicle/battery.py @@ -145,6 +145,9 @@ class Battery(VehicleDataBase): remaining_range_at_full_charge: Optional[ValueWithUnit] = ValueWithUnit(None, None) """Remaining range at full charge of the vehicle.""" + remaining_range_at_20_percent: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """Remaining range when battery reaches 20% SoC.""" + remaining_battery_percent: Optional[ValueWithUnit] = ValueWithUnit(None, None) """Remaining battery percent of the vehicle.""" @@ -178,6 +181,48 @@ class Battery(VehicleDataBase): average_power_consumption: Optional[ValueWithUnit] = ValueWithUnit(None, "W") """Current average consumption""" + # V2L (Vehicle-to-Load) discharge data + v2l_discharge_voltage: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """V2L discharge voltage (disChargeUAct).""" + + v2l_discharge_current: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """V2L discharge current (disChargeIAct).""" + + v2l_discharge_power: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """V2L discharge power (calculated from voltage * current).""" + + v2l_connection_status: Optional[int] = None + """V2L connection status (disChargeConnectStatus). 0=not connected.""" + + v2l_time_remaining: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """Time to target discharge (timeToTargetDisCharged). 2047=not discharging.""" + + # Charge port status + charge_lid_ac_status: Optional[int] = None + """AC charge lid status (chargeLidAcStatus). 2=closed.""" + + charge_lid_dc_status: Optional[int] = None + """DC charge lid status (chargeLidDcAcStatus). 2=closed.""" + + # Scheduled charging + scheduled_charging_status: Optional[int] = None + """Scheduled/book charging status (bookChargeSts). 0=off.""" + + # DC-DC converter status + dcdc_activated: Optional[int] = None + """DC-DC converter activated status (dcDcActvd). 0=not active.""" + + dcdc_connection_status: Optional[int] = None + """DC-DC connection status (dcDcConnectStatus). 0=not connected.""" + + # Wireless charging + wireless_charging_alignment: Optional[int] = None + """Wireless power transfer fine alignment status (wptFineAlignt). 0=not aligned.""" + + # Instantaneous power consumption + instant_power_consumption: Optional[ValueWithUnit] = ValueWithUnit(None, None) + """Instantaneous power consumption (indPowerConsumption).""" + @classmethod def from_vehicle_data(cls, vehicle_data: dict): """Create a new instance based on data from API.""" @@ -263,6 +308,62 @@ def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: if avg_consumption is not None: retval["average_power_consumption"] = ValueWithUnit(avg_consumption, "W") + # Range at 20% SoC + range_20 = get_field_as_type(evStatus, "distanceToEmptyOnBattery20Soc", int, log_missing=False) + if range_20 is not None: + retval["remaining_range_at_20_percent"] = ValueWithUnit(range_20, "km") + + # V2L (Vehicle-to-Load) discharge data + discharge_u = get_field_as_type(evStatus, "disChargeUAct", float, log_missing=False) + discharge_i = get_field_as_type(evStatus, "disChargeIAct", float, log_missing=False) + if discharge_u is not None: + retval["v2l_discharge_voltage"] = ValueWithUnit(discharge_u, "V") + if discharge_i is not None: + retval["v2l_discharge_current"] = ValueWithUnit(discharge_i, "A") + if discharge_u is not None and discharge_i is not None and discharge_u > 0 and discharge_i > 0: + retval["v2l_discharge_power"] = ValueWithUnit(discharge_u * discharge_i, "W") + + discharge_conn = get_field_as_type(evStatus, "disChargeConnectStatus", int, log_missing=False) + if discharge_conn is not None: + retval["v2l_connection_status"] = discharge_conn + + time_to_discharged = get_field_as_type(evStatus, "timeToTargetDisCharged", int, log_missing=False) + if time_to_discharged is not None and time_to_discharged != 2047: + retval["v2l_time_remaining"] = ValueWithUnit(time_to_discharged, "min") + + # Charge lid status + charge_lid_ac = get_field_as_type(evStatus, "chargeLidAcStatus", int, log_missing=False) + if charge_lid_ac is not None: + retval["charge_lid_ac_status"] = charge_lid_ac + + charge_lid_dc = get_field_as_type(evStatus, "chargeLidDcAcStatus", int, log_missing=False) + if charge_lid_dc is not None: + retval["charge_lid_dc_status"] = charge_lid_dc + + # Scheduled charging + book_charge = get_field_as_type(evStatus, "bookChargeSts", int, log_missing=False) + if book_charge is not None: + retval["scheduled_charging_status"] = book_charge + + # DC-DC converter status + dcdc_active = get_field_as_type(evStatus, "dcDcActvd", int, log_missing=False) + if dcdc_active is not None: + retval["dcdc_activated"] = dcdc_active + + dcdc_conn = get_field_as_type(evStatus, "dcDcConnectStatus", int, log_missing=False) + if dcdc_conn is not None: + retval["dcdc_connection_status"] = dcdc_conn + + # Wireless charging alignment + wpt_align = get_field_as_type(evStatus, "wptFineAlignt", int, log_missing=False) + if wpt_align is not None: + retval["wireless_charging_alignment"] = wpt_align + + # Instantaneous power consumption + instant_consumption = get_field_as_type(evStatus, "indPowerConsumption", float, log_missing=False) + if instant_consumption is not None: + retval["instant_power_consumption"] = ValueWithUnit(instant_consumption, "W") + if "vehicleStatus" in vehicle_data and "updateTime" in vehicle_data["vehicleStatus"]: retval["timestamp"] = datetime.fromtimestamp(int(vehicle_data["vehicleStatus"]["updateTime"]) / 1000) diff --git a/pysmarthashtag/vehicle/climate.py b/pysmarthashtag/vehicle/climate.py index d76c9a9..d01ee47 100644 --- a/pysmarthashtag/vehicle/climate.py +++ b/pysmarthashtag/vehicle/climate.py @@ -134,6 +134,15 @@ class Climate(VehicleDataBase): interior_PM25: Optional[ValueWithUnit] = ValueWithUnit(None, None) """The interior PM2.5 value.""" + interior_PM25_level: Optional[int] = None + """The interior PM2.5 air quality level (0-5 scale).""" + + interior_secondary_PM25_level: Optional[int] = None + """The interior secondary PM2.5 level (interiorSecondPM25Level).""" + + exterior_PM25_level: Optional[int] = None + """The exterior PM2.5 air quality level (0-5 scale).""" + relative_humidity: Optional[ValueWithUnit] = ValueWithUnit(None, None) """The relative humidity.""" @@ -205,6 +214,11 @@ def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: retval["interior_PM25"] = ValueWithUnit(interior_pm25, "μg/m³") if interior_pm25 is not None else None rel_hum = get_field_as_type(pollutionStatus, "relHumSts", float) retval["relative_humidity"] = ValueWithUnit(rel_hum, "%") if rel_hum is not None else None + retval["interior_PM25_level"] = get_field_as_type(pollutionStatus, "interiorPM25Level", int) + retval["interior_secondary_PM25_level"] = get_field_as_type( + pollutionStatus, "interiorSecondPM25Level", int + ) + retval["exterior_PM25_level"] = get_field_as_type(pollutionStatus, "exteriorPM25Level", int) retval["timestamp"] = datetime.fromtimestamp(int(vehicle_data["vehicleStatus"]["updateTime"]) / 1000) except KeyError as e: diff --git a/pysmarthashtag/vehicle/safety.py b/pysmarthashtag/vehicle/safety.py index 5e13c9d..a7354ff 100644 --- a/pysmarthashtag/vehicle/safety.py +++ b/pysmarthashtag/vehicle/safety.py @@ -123,9 +123,11 @@ def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: retval["door_open_status_passenger"] = get_field_as_type(evStatus, "doorOpenStatusPassenger", int) retval["door_open_status_passenger_rear"] = get_field_as_type(evStatus, "doorOpenStatusPassengerRear", int) retval["door_pos_driver"] = get_field_as_type(evStatus, "doorPosDriver", int) - retval["door_pos_driver_rear"] = get_field_as_type(evStatus, "doorPosDriverRear", int) + retval["door_pos_driver_rear"] = get_field_as_type(evStatus, "doorPosDriverRear", int, log_missing=False) retval["door_pos_passenger"] = get_field_as_type(evStatus, "doorPosPassenger", int) - retval["door_pos_passenger_rear"] = get_field_as_type(evStatus, "doorPosPassengerRear", int) + retval["door_pos_passenger_rear"] = get_field_as_type( + evStatus, "doorPosPassengerRear", int, log_missing=False + ) retval["electric_park_brake_status"] = get_field_as_type(evStatus, "electricParkBrakeStatus", int) retval["engine_hood_open_status"] = get_field_as_type(evStatus, "engineHoodOpenStatus", int) retval["seat_belt_status_driver"] = get_field_as_type(evStatus, "seatBeltStatusDriver", bool) diff --git a/pysmarthashtag/vehicle/trailer.py b/pysmarthashtag/vehicle/trailer.py new file mode 100644 index 0000000..0944083 --- /dev/null +++ b/pysmarthashtag/vehicle/trailer.py @@ -0,0 +1,68 @@ +"""Trailer status models for pysmarthashtag.""" + +import logging +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Optional + +from pysmarthashtag.models import VehicleDataBase, get_field_as_type + +_LOGGER = logging.getLogger(__name__) + + +@dataclass +class Trailer(VehicleDataBase): + """Provides an accessible version of the vehicle's trailer status data. + + This includes all trailer lamp status information from the trailerStatus + section of the API response. + """ + + turning_lamp_status: Optional[int] = None + """Trailer turning lamp status. 0=off.""" + + fog_lamp_status: Optional[int] = None + """Trailer fog lamp status. 0=off.""" + + brake_lamp_status: Optional[int] = None + """Trailer brake lamp status. 0=off.""" + + reversing_lamp_status: Optional[int] = None + """Trailer reversing lamp status. 0=off.""" + + position_lamp_status: Optional[int] = None + """Trailer position lamp status. 0=off.""" + + @classmethod + def from_vehicle_data(cls, vehicle_data: dict): + """Create a new instance based on data from API.""" + parsed = cls._parse_vehicle_data(vehicle_data) or {} + if len(parsed) > 0: + return cls(**parsed) + return None + + @classmethod + def _parse_vehicle_data(cls, vehicle_data: dict) -> Optional[dict]: + """Parse the trailer status data based on the vehicle data dict.""" + _LOGGER.debug("Parsing trailer status data") + if "vehicleStatus" not in vehicle_data: + return None + retval: dict[str, Any] = {} + try: + trailer_status = vehicle_data["vehicleStatus"]["additionalVehicleStatus"].get("trailerStatus", {}) + if not trailer_status: + return None + + retval["turning_lamp_status"] = get_field_as_type(trailer_status, "trailerTurningLampSts", int) + retval["fog_lamp_status"] = get_field_as_type(trailer_status, "trailerFogLampSts", int) + retval["brake_lamp_status"] = get_field_as_type(trailer_status, "trailerBreakLampSts", int) + retval["reversing_lamp_status"] = get_field_as_type(trailer_status, "trailerReversingLampSts", int) + retval["position_lamp_status"] = get_field_as_type(trailer_status, "trailerPosLampSts", int) + + # Timestamp + if "updateTime" in vehicle_data["vehicleStatus"]: + retval["timestamp"] = datetime.fromtimestamp(int(vehicle_data["vehicleStatus"]["updateTime"]) / 1000) + + except KeyError as e: + _LOGGER.info(f"Trailer status info not available: {e}") + return retval diff --git a/pysmarthashtag/vehicle/vehicle.py b/pysmarthashtag/vehicle/vehicle.py index f632a01..27ffe79 100644 --- a/pysmarthashtag/vehicle/vehicle.py +++ b/pysmarthashtag/vehicle/vehicle.py @@ -6,6 +6,7 @@ from pysmarthashtag.const import API_BASE_URL, API_BASE_URL_V2 from pysmarthashtag.models import ValueWithUnit, get_element_from_dict_maybe +from pysmarthashtag.vehicle.basic_status import BasicStatus from pysmarthashtag.vehicle.battery import Battery from pysmarthashtag.vehicle.climate import Climate from pysmarthashtag.vehicle.maintenance import Maintenance @@ -13,6 +14,7 @@ from pysmarthashtag.vehicle.running import Running from pysmarthashtag.vehicle.safety import Safety from pysmarthashtag.vehicle.tires import Tires +from pysmarthashtag.vehicle.trailer import Trailer _LOGGER = logging.getLogger(__name__) @@ -56,6 +58,12 @@ class SmartVehicle: safety: Optional[Safety] = None """The safety status of the vehicle.""" + basic_status: Optional[BasicStatus] = None + """The basic status of the vehicle (speed, engine, eGuard, park time, etc.).""" + + trailer: Optional[Trailer] = None + """The trailer status of the vehicle.""" + climate_control: Optional["ClimateControll"] = None # noqa: F821 charging_control: Optional["ChargingControl"] = None # noqa: F821 @@ -121,6 +129,8 @@ def combine_data( self.running = Running.from_vehicle_data(self.data) self.climate = Climate.from_vehicle_data(self.data) self.safety = Safety.from_vehicle_data(self.data) + self.basic_status = BasicStatus.from_vehicle_data(self.data) + self.trailer = Trailer.from_vehicle_data(self.data) from pysmarthashtag.control.climate import ClimateControll