m/hacs-oasis_mini
1
0
Fork 0
mirror of https://github.com/natekspencer/hacs-oasis_mini.git synced 2025-12-06 18:44:14 -05:00
Code Wiki Activity

Ensure update_tracks closes the connection

Browse Source
This commit is contained in:
Nathan Spencer
2025-11-24 01:26:46 +00:00
parent 0ebab392fb
commit 0df118d18d
1 changed files with 40 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
update_tracks.py
View File

@@ -16,11 +16,11 @@ ACCESS_TOKEN = os.getenv("GROUNDED_TOKEN")
def get_author_name(data: dict) -> str: def get_author_name(data: dict) -> str:
""" """
Extracts the author's display name from a nested track data dictionary. Extracts the author's display name from a nested track data dictionary.
Parameters: Parameters:
data (dict): A mapping representing track/result data. Expected shapes include data (dict): A mapping representing track/result data. Expected shapes include
{"author": {"user": {"name": ..., "nickname": ...}}} or {"author": {"name": ..., "nickname": ...}}. {"author": {"user": {"name": ..., "nickname": ...}}} or {"author": {"name": ..., "nickname": ...}}.
Returns: Returns:
str: The author's `name` if present, otherwise the author's `nickname`, otherwise "Kinetic Oasis". str: The author's `name` if present, otherwise the author's `nickname`, otherwise "Kinetic Oasis".
""" """
@@ -30,47 +30,51 @@ def get_author_name(data: dict) -> str:
async def update_tracks() -> None: async def update_tracks() -> None:
""" """
Fetch tracks from the Grounded Labs cloud, detect new or changed public tracks compared to the local TRACKS mapping, augment changed entries with author and reduced SVG content, and persist the merged, sorted track list to custom_components/oasis_mini/pyoasiscontrol/tracks.json. Fetch tracks from the Grounded Labs cloud, detect new or changed public tracks
compared to the local TRACKS mapping, augment changed entries with author and
reduced SVG content, and persist the merged, sorted track list to
`custom_components/oasis_mini/pyoasiscontrol/tracks.json`.
Side effects: Side effects:
- May print error or status messages to stdout. - May print error or status messages to stdout.
- Writes the updated tracks JSON file. - Writes the updated tracks JSON file.
- Ensures the OasisCloudClient session is closed and returns early on errors or unexpected data. - Ensures the OasisCloudClient session is closed and returns early on errors or
unexpected data.
""" """
client = OasisCloudClient(access_token=ACCESS_TOKEN) client = OasisCloudClient(access_token=ACCESS_TOKEN)
try: try:
data = await client.async_get_tracks() try:
except Exception as ex: data = await client.async_get_tracks()
print(type(ex).__name__, ex) except Exception as ex:
await client.session.close() print(type(ex).__name__, ex)
return return
if not isinstance(data, list): if not isinstance(data, list):
print("Unexpected result:", data) print("Unexpected result:", data)
return return
updated_tracks: dict[int, dict[str, Any]] = {} updated_tracks: dict[int, dict[str, Any]] = {}
for result in filter(lambda d: d["public"], data): for result in filter(lambda d: d["public"], data):
if ( if (
(track_id := result["id"]) not in TRACKS (track_id := result["id"]) not in TRACKS
or any( or any(
result[field] != TRACKS[track_id].get(field) result[field] != TRACKS[track_id].get(field)
for field in ("name", "image", "png_image") for field in ("name", "image", "png_image")
) )
or TRACKS[track_id].get("author") != get_author_name(result) or TRACKS[track_id].get("author") != get_author_name(result)
): ):
print(f"Updating track {track_id}: {result['name']}") print(f"Updating track {track_id}: {result['name']}")
track_info = await client.async_get_track_info(int(track_id)) track_info = await client.async_get_track_info(int(track_id))
if not track_info: if not track_info:
print("No track info") print("No track info")
break break
result["author"] = get_author_name(result) result["author"] = get_author_name(result)
result["reduced_svg_content_new"] = track_info.get( result["reduced_svg_content_new"] = track_info.get(
"reduced_svg_content_new" "reduced_svg_content_new"
) )
updated_tracks[track_id] = result updated_tracks[track_id] = result
await client.session.close() finally:
await client.async_close()
if not updated_tracks: if not updated_tracks:
print("No updated tracks") print("No updated tracks")
@@ -87,4 +91,4 @@ async def update_tracks() -> None:
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run(update_tracks()) asyncio.run(update_tracks())