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

📝 Add docstrings to mqtt

Browse Source 
Docstrings generation was requested by @natekspencer.

* https://github.com/natekspencer/hacs-oasis_mini/pull/98#issuecomment-3568450288

The following files were modified:

* `custom_components/oasis_mini/__init__.py`
* `custom_components/oasis_mini/binary_sensor.py`
* `custom_components/oasis_mini/button.py`
* `custom_components/oasis_mini/config_flow.py`
* `custom_components/oasis_mini/coordinator.py`
* `custom_components/oasis_mini/entity.py`
* `custom_components/oasis_mini/helpers.py`
* `custom_components/oasis_mini/image.py`
* `custom_components/oasis_mini/light.py`
* `custom_components/oasis_mini/media_player.py`
* `custom_components/oasis_mini/number.py`
* `custom_components/oasis_mini/pyoasiscontrol/clients/cloud_client.py`
* `custom_components/oasis_mini/pyoasiscontrol/clients/http_client.py`
* `custom_components/oasis_mini/pyoasiscontrol/clients/mqtt_client.py`
* `custom_components/oasis_mini/pyoasiscontrol/clients/transport.py`
* `custom_components/oasis_mini/pyoasiscontrol/device.py`
* `custom_components/oasis_mini/pyoasiscontrol/utils.py`
* `custom_components/oasis_mini/select.py`
* `custom_components/oasis_mini/sensor.py`
* `custom_components/oasis_mini/switch.py`
* `custom_components/oasis_mini/update.py`
* `update_tracks.py`
This commit is contained in:
coderabbitai[bot]
2025-11-23 23:18:59 +00:00
committed by GitHub
parent cf21a5d995
commit 4ef28fc741
22 changed files with 1635 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

335
custom_components/oasis_mini/pyoasiscontrol/device.py
View File

@@ -70,6 +70,23 @@ class OasisDevice:
client: OasisClientProtocol | None = None,
) -> None:
# Transport
"""
Initialize an OasisDevice with identification, network, transport references, and default state fields.
Parameters:
model (str | None): Device model identifier.
serial_number (str | None): Device serial number.
name (str | None): Human-readable device name; if omitted, defaults to "<model> <serial_number>".
ssid (str | None): Last-known Wi‑Fi SSID for the device.
ip_address (str | None): Last-known IP address for the device.
cloud (OasisCloudClient | None): Optional cloud client used to fetch track metadata and remote data.
client (OasisClientProtocol | None): Optional transport client used to send commands to the device.
Notes:
- Creates an internal listener list for state-change callbacks.
- Initializes status fields (brightness, playlist, playback state, networking, etc.) with sensible defaults.
- Initializes a track metadata cache and a placeholder for a background refresh task.
"""
self._cloud = cloud
self._client = client
self._listeners: list[Callable[[], None]] = []
@@ -118,18 +135,34 @@ class OasisDevice:
@property
def brightness(self) -> int:
"""Return the brightness."""
"""
Current display brightness adjusted for the device sleep state.
Returns:
int: 0 when the device is sleeping, otherwise the stored brightness value.
"""
return 0 if self.is_sleeping else self._brightness
@brightness.setter
def brightness(self, value: int) -> None:
"""
Set the device brightness and update brightness_on when non-zero.
Parameters:
value (int): Brightness level to apply; if non-zero, also stored in `brightness_on`.
"""
self._brightness = value
if value:
self.brightness_on = value
@property
def is_sleeping(self) -> bool:
"""Return `True` if the status is set to sleeping."""
"""
Indicates whether the device is currently in the sleeping status.
Returns:
`true` if the device is sleeping, `false` otherwise.
"""
return self.status_code == STATUS_CODE_SLEEPING
def attach_client(self, client: OasisClientProtocol) -> None:
@@ -138,11 +171,24 @@ class OasisDevice:
@property
def client(self) -> OasisClientProtocol | None:
"""Return the current transport client, if any."""
"""
Get the attached transport client, or `None` if no client is attached.
Returns:
The attached transport client, or `None` if not attached.
"""
return self._client
def _require_client(self) -> OasisClientProtocol:
"""Return the attached client or raise if missing."""
"""
Get the attached transport client for this device.
Returns:
OasisClientProtocol: The attached transport client.
Raises:
RuntimeError: If no client/transport is attached to the device.
"""
if self._client is None:
raise RuntimeError(
f"No client/transport attached for device {self.serial_number!r}"
@@ -150,6 +196,18 @@ class OasisDevice:
return self._client
def _update_field(self, name: str, value: Any) -> bool:
"""
Update an attribute on the device if the new value differs from the current value.
Sets the instance attribute named `name` to `value` and logs a debug message when a change occurs.
Parameters:
name (str): The attribute name to update.
value (Any): The new value to assign to the attribute.
Returns:
bool: True if the attribute was changed, False otherwise.
"""
old = getattr(self, name, None)
if old != value:
_LOGGER.debug(
@@ -164,7 +222,13 @@ class OasisDevice:
return False
def update_from_status_dict(self, data: dict[str, Any]) -> None:
"""Update device fields from a status payload (from any transport)."""
"""
Update the device's attributes from a status dictionary.
Expects a mapping of attribute names to values; known attributes are applied to the device,
unknown keys are logged and ignored. If `playlist` or `playlist_index` change, a track
refresh is scheduled. If any attribute changed, registered update listeners are notified.
"""
changed = False
playlist_or_index_changed = False
@@ -184,11 +248,22 @@ class OasisDevice:
self._notify_listeners()
def parse_status_string(self, raw_status: str) -> dict[str, Any] | None:
"""Parse a semicolon-separated status string into a state dict.
Used by:
- HTTP GETSTATUS response
- MQTT FULLSTATUS payload (includes software_version)
"""
Parse a semicolon-separated device status string into a structured state dictionary.
Expects a semicolon-separated string containing at least 18 fields (device status format returned by the device: e.g., HTTP GETSTATUS or MQTT FULLSTATUS). Returns None for empty input or if the string cannot be parsed into the expected fields.
Parameters:
raw_status (str): Semicolon-separated status string from the device.
Returns:
dict[str, Any] | None: A dictionary with these keys on success:
- `status_code`, `error`, `ball_speed`, `playlist` (list[int]), `playlist_index`,
`progress`, `led_effect`, `led_color_id`, `led_speed`, `brightness`, `color`,
`busy`, `download_progress`, `brightness_max`, `wifi_connected`, `repeat_playlist`,
`autoplay`, `auto_clean`
- `software_version` (str) is included if an additional trailing field is present.
Returns `None` if the input is empty or parsing fails.
"""
if not raw_status:
return None
@@ -239,35 +314,71 @@ class OasisDevice:
return status
def update_from_status_string(self, raw_status: str) -> None:
"""Parse and apply a raw status string."""
"""
Parse a semicolon-separated device status string and apply the resulting fields to the device state.
If the string cannot be parsed into a valid status dictionary, no state is changed.
Parameters:
raw_status (str): Raw status payload received from the device (semicolon-separated fields).
"""
if status := self.parse_status_string(raw_status):
self.update_from_status_dict(status)
def as_dict(self) -> dict[str, Any]:
"""Return core state as a dict."""
"""
Return a mapping of the device's core state fields to their current values.
Returns:
dict[str, Any]: A dictionary whose keys are the core state field names (as defined in _STATE_FIELDS)
and whose values are the current values for those fields.
"""
return {field: getattr(self, field) for field in _STATE_FIELDS}
@property
def error_message(self) -> str | None:
"""Return the error message, if any."""
"""
Get the human-readable error message for the current device error code.
Returns:
str: The mapped error message when the device status indicates an error (status code 9); `None` otherwise.
"""
if self.status_code == 9:
return ERROR_CODE_MAP.get(self.error, f"Unknown ({self.error})")
return None
@property
def status(self) -> str:
"""Return human-readable status from status_code."""
"""
Get a human-readable status description for the current status_code.
Returns:
str: Human-readable status corresponding to the device's status_code, or "Unknown (<code>)" when the code is not recognized.
"""
return STATUS_CODE_MAP.get(self.status_code, f"Unknown ({self.status_code})")
@property
def track(self) -> dict | None:
"""Return cached track info if it matches the current `track_id`."""
"""
Return the cached track metadata when it corresponds to the current track, otherwise retrieve the built-in track metadata.
Returns:
dict | None: The track metadata dictionary for the current `track_id`, or `None` if no matching track is available.
"""
if (track := self._track) and track["id"] == self.track_id:
return track
return TRACKS.get(self.track_id)
@property
def track_id(self) -> int | None:
"""
Determine the current track id from the active playlist.
If the playlist index is beyond the end of the playlist, the first track id is returned.
Returns:
int | None: The current track id, or `None` if there is no playlist.
"""
if not self.playlist:
return None
i = self.playlist_index
@@ -275,21 +386,40 @@ class OasisDevice:
@property
def track_image_url(self) -> str | None:
"""Return the track image url, if any."""
"""
Get the full HTTPS URL for the current track's image if available.
Returns:
str: Full URL to the track image (https://app.grounded.so/uploads/<image>), or `None` if no image is available.
"""
if (track := self.track) and (image := track.get("image")):
return f"https://app.grounded.so/uploads/{image}"
return None
@property
def track_name(self) -> str | None:
"""Return the track name, if any."""
"""
Get the current track's display name.
If the current track has no explicit name, returns "Unknown Title (#{track_id})". If there is no current track, returns None.
Returns:
str | None: The track name, or `None` if no current track is available.
"""
if track := self.track:
return track.get("name", f"Unknown Title (#{self.track_id})")
return None
@property
def drawing_progress(self) -> float | None:
"""Return drawing progress percentage for the current track."""
"""
Compute drawing progress percentage for the current track.
If the current track or its SVG content is unavailable, returns None.
Returns:
progress_percent (float | None): Percentage of the drawing completed (0–100), clamped to 100; `None` if no track or SVG content is available.
"""
if not (self.track and (svg_content := self.track.get("svg_content"))):
return None
svg_content = decrypt_svg_content(svg_content)
@@ -300,7 +430,12 @@ class OasisDevice:
@property
def playlist_details(self) -> dict[int, dict[str, str]]:
"""Basic playlist details using built-in TRACKS metadata."""
"""
Build a mapping of track IDs in the current playlist to their detail dictionaries, preferring the device's cached/current track data and falling back to built-in TRACKS.
Returns:
dict[int, dict[str, str]]: A mapping from track ID to a details dictionary (contains at least a `'name'` key). If track metadata is available from the device cache or built-in TRACKS it is used; otherwise a fallback `{"name": "Unknown Title (#<id>)"}` is provided.
"""
return {
track_id: {self.track_id: self.track or {}, **TRACKS}.get(
track_id,
@@ -310,17 +445,32 @@ class OasisDevice:
}
def create_svg(self) -> str | None:
"""Create the current svg based on track and progress."""
"""
Generate an SVG representing the current track at the device's drawing progress.
Returns:
svg (str | None): SVG content for the current track reflecting current progress, or None if track data is unavailable.
"""
return create_svg(self.track, self.progress)
def add_update_listener(self, listener: Callable[[], None]) -> Callable[[], None]:
"""Register a callback for state changes.
Returns an unsubscribe function.
"""
Register a callback to be invoked whenever the device state changes.
Parameters:
listener (Callable[[], None]): A zero-argument callback that will be called on state updates.
Returns:
Callable[[], None]: An unsubscribe function that removes the registered listener; calling the unsubscribe function multiple times is safe.
"""
self._listeners.append(listener)
def _unsub() -> None:
"""
Remove the previously registered listener from the device's listener list if it is present.
This function silently does nothing if the listener is not found.
"""
try:
self._listeners.remove(listener)
except ValueError:
@@ -329,7 +479,11 @@ class OasisDevice:
return _unsub
def _notify_listeners(self) -> None:
"""Call all registered listeners."""
"""
Invoke all registered update listeners in registration order.
Each listener is called synchronously; exceptions raised by a listener are caught and logged so other listeners still run.
"""
for listener in list(self._listeners):
try:
listener()
@@ -337,7 +491,12 @@ class OasisDevice:
_LOGGER.exception("Error in update listener")
async def async_get_mac_address(self) -> str | None:
"""Return the device MAC address, refreshing via transport if needed."""
"""
Get the device MAC address, requesting it from the attached transport client if not already known.
Returns:
mac (str | None): The device MAC address if available, otherwise `None`.
"""
if self.mac_address:
return self.mac_address
@@ -348,10 +507,25 @@ class OasisDevice:
return mac
async def async_set_auto_clean(self, auto_clean: bool) -> None:
"""
Set whether the device performs automatic cleaning.
Parameters:
auto_clean (bool): `True` to enable automatic cleaning, `False` to disable it.
"""
client = self._require_client()
await client.async_send_auto_clean_command(self, auto_clean)
async def async_set_ball_speed(self, speed: int) -> None:
"""
Set the device's ball speed.
Parameters:
speed (int): Desired ball speed in the allowed range (BALL_SPEED_MIN to BALL_SPEED_MAX, inclusive).
Raises:
ValueError: If `speed` is outside the allowed range.
"""
if not BALL_SPEED_MIN <= speed <= BALL_SPEED_MAX:
raise ValueError("Invalid speed specified")
client = self._require_client()
@@ -365,7 +539,19 @@ class OasisDevice:
led_speed: int | None = None,
brightness: int | None = None,
) -> None:
"""Set the Oasis device LED (shared validation & attribute updates)."""
"""
Set the device LED effect, color, speed, and brightness.
Parameters:
led_effect (str | None): LED effect name; if None, the device's current effect is used. Must be one of the supported LED effects.
color (str | None): Hex color string (e.g. "#rrggbb"); if None, the device's current color is used or `#ffffff` if unset.
led_speed (int | None): LED animation speed; if None, the device's current speed is used. Must be within the allowed LED speed range.
brightness (int | None): Brightness level; if None, the device's current brightness is used. Must be between 0 and the device's `brightness_max`.
Raises:
ValueError: If `led_effect` is not supported, or `led_speed` or `brightness` are outside their valid ranges.
RuntimeError: If no transport client is attached to the device.
"""
if led_effect is None:
led_effect = self.led_effect
if color is None:
@@ -388,18 +574,48 @@ class OasisDevice:
)
async def async_sleep(self) -> None:
"""
Put the device into sleep mode.
Sends a sleep command to the attached transport client.
Raises:
RuntimeError: If no client is attached.
"""
client = self._require_client()
await client.async_send_sleep_command(self)
async def async_move_track(self, from_index: int, to_index: int) -> None:
"""
Move a track within the device's playlist from one index to another.
Parameters:
from_index (int): Index of the track to move within the current playlist.
to_index (int): Destination index where the track should be placed.
"""
client = self._require_client()
await client.async_send_move_job_command(self, from_index, to_index)
async def async_change_track(self, index: int) -> None:
"""
Change the device's current track to the track at the given playlist index.
Parameters:
index (int): Zero-based index of the track in the device's current playlist.
"""
client = self._require_client()
await client.async_send_change_track_command(self, index)
async def async_add_track_to_playlist(self, track: int | Iterable[int]) -> None:
"""
Add one or more tracks to the device's playlist via the attached client.
Parameters:
track (int | Iterable[int]): A single track id or an iterable of track ids to append to the playlist.
Raises:
RuntimeError: If no transport client is attached to the device.
"""
if isinstance(track, int):
tracks = [track]
else:
@@ -408,6 +624,14 @@ class OasisDevice:
await client.async_send_add_joblist_command(self, tracks)
async def async_set_playlist(self, playlist: int | Iterable[int]) -> None:
"""
Set the device's playlist to the provided track or tracks.
Accepts a single track ID or an iterable of track IDs and replaces the device's playlist by sending the corresponding command to the attached client.
Parameters:
playlist (int | Iterable[int]): A single track ID or an iterable of track IDs to set as the new playlist.
"""
if isinstance(playlist, int):
playlist_list = [playlist]
else:
@@ -416,38 +640,85 @@ class OasisDevice:
await client.async_send_set_playlist_command(self, playlist_list)
async def async_set_repeat_playlist(self, repeat: bool) -> None:
"""
Set whether the device's playlist should repeat.
Parameters:
repeat (bool): True to enable repeating the playlist, False to disable it.
"""
client = self._require_client()
await client.async_send_set_repeat_playlist_command(self, repeat)
async def async_set_autoplay(self, option: bool | int | str) -> None:
"""Set autoplay / wait-after behavior."""
"""
Set the device's autoplay / wait-after option.
Parameters:
option (bool | int | str): Desired autoplay/wait-after value. If a `bool` is provided, `True` is converted to `"0"` and `False` to `"1"`. Integer or string values are sent as their string representation.
"""
if isinstance(option, bool):
option = 0 if option else 1
client = self._require_client()
await client.async_send_set_autoplay_command(self, str(option))
async def async_upgrade(self, beta: bool = False) -> None:
"""
Initiates a firmware upgrade on the device.
Parameters:
beta (bool): If True, request a beta (pre-release) firmware; otherwise request the stable firmware.
"""
client = self._require_client()
await client.async_send_upgrade_command(self, beta)
async def async_play(self) -> None:
"""
Send a play command to the device via the attached transport client.
Raises:
RuntimeError: If no transport client is attached.
"""
client = self._require_client()
await client.async_send_play_command(self)
async def async_pause(self) -> None:
"""
Pause playback on the device.
Raises:
RuntimeError: If no transport client is attached.
"""
client = self._require_client()
await client.async_send_pause_command(self)
async def async_stop(self) -> None:
"""
Stop playback on the device by sending a stop command through the attached transport client.
Raises:
RuntimeError: if no transport client is attached to the device.
"""
client = self._require_client()
await client.async_send_stop_command(self)
async def async_reboot(self) -> None:
"""
Reboots the device using the attached transport client.
Requests the attached client to send a reboot command to the device.
Raises:
RuntimeError: If no transport client is attached.
"""
client = self._require_client()
await client.async_send_reboot_command(self)
def schedule_track_refresh(self) -> None:
"""Schedule an async refresh of current track info if track_id changed."""
"""
Schedule a background refresh of the current track metadata when the device's track may have changed.
Does nothing if no cloud client is attached or if there is no running event loop. If a previous refresh task is still pending, it is cancelled before a new background task is scheduled.
"""
if not self._cloud:
return
@@ -463,7 +734,11 @@ class OasisDevice:
self._track_task = loop.create_task(self._async_refresh_current_track())
async def _async_refresh_current_track(self) -> None:
"""Refresh the current track info."""
"""
Refresh cached information for the current track by fetching details from the attached cloud client and notify listeners when updated.
If no cloud client is attached, no current track exists, or the cached track already matches the current track id, the method returns without change. On successful fetch, updates the device's track cache and invokes registered update listeners.
"""
if not self._cloud:
return
@@ -484,4 +759,4 @@ class OasisDevice:
return
self._track = track
self._notify_listeners()
self._notify_listeners()