[seaspotter]

UI: openWB/openwb-ui-settings#1153

Summary by Sourcery

Add Pleos API support for retrieving and reporting Hyundai, Kia, and Genesis vehicle status data.

New Features:

• Add Pleos vehicle integration for Hyundai, Kia, and Genesis vehicles using the EU Data Act API.
• Expose battery state of charge, driving range, odometer, charging-limit warnings, and timestamped vehicle data through the configurable vehicle framework.

Bug Fixes:

• Redact client secrets in logged sensitive data.

Enhancements:

• Handle Pleos token reuse, expiration, and one-time refresh after unauthorized responses, with unit conversion and timestamp validation for vehicle data.

Tests:

• Add coverage for Pleos API authentication, vehicle data conversion, error handling, timestamp validation, and token refresh behavior.

[LKuemmel]

@sourcery-ai review

[sourcery-ai]

Hey - I've found 8 issues

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="packages/modules/vehicles/pleos/config.py" line_range="18" />
<code_context>
+    def __init__(self,
+                 brand: Optional[str] = None,
+                 client_id: Optional[str] = None,
+                 client_secret: Optional[str] = None,
+                 vin: Optional[str] = None,
+                 token: Optional[PleosToken] = None) -> None:
</code_context>
<issue_to_address>
**Client credentials enter MQTT logs**

When an operator configures Pleos and submits the `soc_module/config` MQTT payload, `SubData.on_message` logs the full configuration payload to `mqtt.log`; its redaction filter matches `secret` as an exact field name, not `client_secret`, so the submitted credential remains readable to anyone who obtains the logs.

Redact `client_secret` and other credential field variants, or avoid logging configuration payloads containing credentials.
</issue_to_address>

### Comment 2
<location path="packages/modules/vehicles/pleos/api.py" line_range="44" />
<code_context>
+
+def get_battery(brand: str, vin: str, token: PleosToken) -> dict:
+    response = req.get_http_session().get(
+        BASE_URL + f"/vehicles/{vin}/batteries", headers=_auth_headers(brand, token), timeout=10)
+    return response.json().get("data", {})
+
</code_context>
<issue_to_address>
**Vehicle telemetry enters logs**

When a user has configured Pleos and its vehicle update polls successfully, `req.get_http_session()` logs each raw response body at DEBUG, and the root logger is enabled at DEBUG, so normal polling records the vehicle's charge level, estimated range, and odometer in system logs for anyone who obtains them.

Do not log raw vehicle API response bodies; log only non-sensitive status or diagnostic metadata.

Also at `packages/modules/vehicles/pleos/api.py:50`, `packages/modules/vehicles/pleos/api.py:59`.
</issue_to_address>

### Comment 3
<location path="packages/modules/vehicles/pleos/api.py" line_range="77" />
<code_context>
+    for entry in response.json().get("data", {}).get("distanceToEmpties", []):
+        if entry.get("type") == "EV":
+            return _to_km(entry.get("value"), entry.get("unit"))
+    return None
+
+
</code_context>
<issue_to_address>
**Stale charge appears current**

When the battery response omits its timestamp or provides one that `parse_timestamp` cannot parse, `parse_timestamp` returns `None`, `CarState` replaces it with the current time, and `ConfigurableVehicle` treats the stale SoC as fresh, overwriting newer data.

Preserve missing or invalid timestamps as invalid, or use a fallback that does not make the reading appear fresh.

Also at `packages/modules/vehicles/pleos/soc.py:22-28`.
</issue_to_address>

### Comment 4
<location path="packages/modules/vehicles/pleos/soc.py" line_range="27-28" />
<code_context>
+    warning = (f"Das Fahrzeug begrenzt die Ladung selbst auf {target_soc}%."
+               if target_soc and target_soc < 100 else None)
+    range = api.get_distance_to_empty_km(config.brand, config.vin, config.token)
+    odometer = api.get_odometer_km(config.brand, config.vin, config.token)
+    return CarState(soc=soc, range=range, soc_timestamp=soc_timestamp, odometer=odometer, warning=warning)
+
+
</code_context>
<issue_to_address>
**Optional data failure drops SoC**

When the range or odometer request fails while the battery request succeeds, `get_distance_to_empty_km` or `get_odometer_km` raises after the battery response succeeds, so `fetch` never returns its valid SoC. `ConfigurableVehicle` then treats the updater as failed, and without its charging fallback the vehicle keeps its previous SoC.

Handle failures from the optional range and odometer requests independently so they do not discard a successfully fetched SoC.
</issue_to_address>

### Comment 5
<location path="packages/modules/vehicles/pleos/api.py" line_range="51" />
<code_context>
+def get_distance_to_empty_km(brand: str, vin: str, token: PleosToken) -> Optional[float]:
+    response = req.get_http_session().get(
+        BASE_URL + f"/vehicles/{vin}/powertrains", headers=_auth_headers(brand, token), timeout=10)
+    for entry in response.json().get("data", {}).get("distanceToEmpties", []):
+        if entry.get("type") == "EV":
+            return _to_km(entry.get("value"), entry.get("unit"))
</code_context>
<issue_to_address>
**Optional data aborts SoC fetch**

When the API returns `null` for `distanceToEmpties`, `odometer`, or `targetStateOfCharge`, a `null` optional value makes these nested `.get()` calls or the range loop raise before `fetch` can return the valid SoC. The whole vehicle update then fails instead of leaving range, odometer, or warning unset.

Normalize nullable nested objects and lists before accessing or iterating them, so unavailable optional fields produce `None` rather than aborting the update.

Also at `packages/modules/vehicles/pleos/api.py:60-61`, `packages/modules/vehicles/pleos/soc.py:23`.
</issue_to_address>

### Comment 6
<location path="packages/modules/vehicles/pleos/api.py" line_range="18-20" />
<code_context>
+
+
+def validate_token(brand: str, client_id: str, client_secret: str, token: PleosToken) -> PleosToken:
+    if token.access_token is not None and time.time() < token.created_at + token.expires_in:
+        return token
+    return _request_token(brand, client_id, client_secret)
+
+
</code_context>
<issue_to_address>
**Revoked token stalls updates**

When pleos invalidates or rejects an access token before its advertised expiry, `validate_token` trusts the local expiry alone, so a token rejected by Pleos before that expiry is reused and each request fails until the cached expiry passes; the vehicle provides no SoC during that period.

On an authorization failure, refresh the token and retry the request once.
</issue_to_address>

### Comment 7
<location path="packages/modules/vehicles/pleos/soc.py" line_range="24-26" />
<code_context>
+    soc = float(charge['stateOfCharge'])
+    soc_timestamp = api.parse_timestamp(battery.get('timestamp'))
+    target_soc = charge.get('targetStateOfCharge', {}).get('standard')
+    warning = (f"Das Fahrzeug begrenzt die Ladung selbst auf {target_soc}%."
+               if target_soc and target_soc < 100 else None)
+    range = api.get_distance_to_empty_km(config.brand, config.vin, config.token)
+    odometer = api.get_odometer_km(config.brand, config.vin, config.token)
+    return CarState(soc=soc, range=range, soc_timestamp=soc_timestamp, odometer=odometer, warning=warning)
</code_context>
<issue_to_address>
**Cleared limit warning remains**

When a Pleos target below 100% was reported previously and a later update reports 100% or no target, `fetch` returns no warning once the target is 100% or missing, but `ConfigurableVehicle` only sets a fault warning when `CarState.warning` is truthy. The earlier Pleos warning therefore remains stored after the vehicle-side limit is removed, so users continue to see a stale warning.

Clear the vehicle warning on a successful update when the returned `CarState` has no warning.
</issue_to_address>

### Comment 8
<location path="packages/modules/vehicles/pleos/api_test.py" line_range="43-46" />
<code_context>
+        monkeypatch.setattr(api.req, "get_http_session", lambda: session)
+
+        try:
+            api.validate_token("kia", "cid", "csec", PleosToken())
+            assert False, "expected Exception"
+        except Exception:
+            pass
+
+
</code_context>
<issue_to_address>
**Missing-token test passes incorrectly**

When `validate_token` returns normally instead of raising for a response with no access token, the broad `except Exception` catches the `AssertionError` from `assert False`, so the test passes without detecting that `validate_token` returned normally.

Use an exception assertion that does not catch the test's own failure, such as `pytest.raises`.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook