jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Allow configuring Starlink sleep schedule (#103057)

Browse Source 
* Expose sleep config getters and setters

* Add a switch for toggling sleep schedule

* Add Time platform

* Add frozen to dataclasses

* Update tests

* Add starlink time to coveragerc

* No more mixin

* Update time.py

* Update time.py

* Run data collectors asynchronously

* Fix timezone handling
This commit is contained in:
Jack Boswell 2024-03-19 04:18:32 +13:00 committed by GitHub
parent e882d47cde
commit 34b0ff40f3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 204 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -1284,6 +1284,7 @@ omit =
homeassistant/components/starlink/device_tracker.py
homeassistant/components/starlink/sensor.py
homeassistant/components/starlink/switch.py
homeassistant/components/starlink/time.py
homeassistant/components/starline/__init__.py
homeassistant/components/starline/account.py
homeassistant/components/starline/binary_sensor.py

1
homeassistant/components/starlink/__init__.py
View File

@ -15,6 +15,7 @@ PLATFORMS = [
Platform.DEVICE_TRACKER,
Platform.SENSOR,
Platform.SWITCH,
Platform.TIME,
]

63
homeassistant/components/starlink/coordinator.py
View File

@ -6,6 +6,7 @@ import asyncio
from dataclasses import dataclass
from datetime import timedelta
import logging
from zoneinfo import ZoneInfo
from starlink_grpc import (
AlertDict,
@ -14,8 +15,10 @@ from starlink_grpc import (
LocationDict,
ObstructionDict,
StatusDict,
get_sleep_config,
location_data,
reboot,
set_sleep_config,
set_stow_state,
status_data,
)
@ -32,6 +35,7 @@ class StarlinkData:
"""Contains data pulled from the Starlink system."""
location: LocationDict
sleep: tuple[int, int, bool]
status: StatusDict
obstruction: ObstructionDict
alert: AlertDict
@ -43,7 +47,7 @@ class StarlinkUpdateCoordinator(DataUpdateCoordinator[StarlinkData]):
def __init__(self, hass: HomeAssistant, name: str, url: str) -> None:
"""Initialize an UpdateCoordinator for a group of sensors."""
self.channel_context = ChannelContext(target=url)
self.timezone = ZoneInfo(hass.config.time_zone)
super().__init__(
hass,
_LOGGER,
@ -54,13 +58,16 @@ class StarlinkUpdateCoordinator(DataUpdateCoordinator[StarlinkData]):
async def _async_update_data(self) -> StarlinkData:
async with asyncio.timeout(4):
try:
status = await self.hass.async_add_executor_job(
status_data, self.channel_context
status, location, sleep = await asyncio.gather(
self.hass.async_add_executor_job(status_data, self.channel_context),
self.hass.async_add_executor_job(
location_data, self.channel_context
),
self.hass.async_add_executor_job(
get_sleep_config, self.channel_context
),
)
location = await self.hass.async_add_executor_job(
location_data, self.channel_context
)
return StarlinkData(location, *status)
return StarlinkData(location, sleep, *status)
except GrpcError as exc:
raise UpdateFailed from exc
@ -81,3 +88,45 @@ class StarlinkUpdateCoordinator(DataUpdateCoordinator[StarlinkData]):
await self.hass.async_add_executor_job(reboot, self.channel_context)
except GrpcError as exc:
raise HomeAssistantError from exc
async def async_set_sleep_schedule_enabled(self, sleep_schedule: bool) -> None:
"""Set whether Starlink system uses the configured sleep schedule."""
async with asyncio.timeout(4):
try:
await self.hass.async_add_executor_job(
set_sleep_config,
self.data.sleep[0],
self.data.sleep[1],
sleep_schedule,
self.channel_context,
)
except GrpcError as exc:
raise HomeAssistantError from exc
async def async_set_sleep_start(self, start: int) -> None:
"""Set Starlink system sleep schedule start time."""
async with asyncio.timeout(4):
try:
await self.hass.async_add_executor_job(
set_sleep_config,
start,
self.data.sleep[1],
self.data.sleep[2],
self.channel_context,
)
except GrpcError as exc:
raise HomeAssistantError from exc
async def async_set_sleep_duration(self, end: int) -> None:
"""Set Starlink system sleep schedule end time."""
async with asyncio.timeout(4):
try:
await self.hass.async_add_executor_job(
set_sleep_config,
self.data.sleep[0],
end,
self.data.sleep[2],
self.channel_context,
)
except GrpcError as exc:
raise HomeAssistantError from exc

11
homeassistant/components/starlink/strings.json
View File

@ -75,6 +75,17 @@
"switch": {
"stowed": {
"name": "Stowed"
},
"sleep_schedule": {
"name": "Sleep schedule"
}
},
"time": {
"sleep_start": {
"name": "Sleep start"
},
"sleep_end": {
"name": "Sleep end"
}
}
}

14
homeassistant/components/starlink/switch.py
View File

@ -67,5 +67,17 @@ SWITCHES = [
value_fn=lambda data: data.status["state"] == "STOWED",
turn_on_fn=lambda coordinator: coordinator.async_stow_starlink(True),
turn_off_fn=lambda coordinator: coordinator.async_stow_starlink(False),
)
),
StarlinkSwitchEntityDescription(
key="sleep_schedule",
translation_key="sleep_schedule",
device_class=SwitchDeviceClass.SWITCH,
value_fn=lambda data: data.sleep[2],
turn_on_fn=lambda coordinator: coordinator.async_set_sleep_schedule_enabled(
True
),
turn_off_fn=lambda coordinator: coordinator.async_set_sleep_schedule_enabled(
False
),
),
]

98
homeassistant/components/starlink/time.py Normal file
View File

@ -0,0 +1,98 @@
"""Contains time pickers exposed by the Starlink integration."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import UTC, datetime, time, tzinfo
import math
from homeassistant.components.time import TimeEntity, TimeEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .coordinator import StarlinkData, StarlinkUpdateCoordinator
from .entity import StarlinkEntity
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up all time entities for this entry."""
coordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
StarlinkTimeEntity(coordinator, description) for description in TIMES
)
@dataclass(frozen=True, kw_only=True)
class StarlinkTimeEntityDescription(TimeEntityDescription):
"""Describes a Starlink time entity."""
value_fn: Callable[[StarlinkData, tzinfo], time | None]
update_fn: Callable[[StarlinkUpdateCoordinator, time], Awaitable[None]]
available_fn: Callable[[StarlinkData], bool]
class StarlinkTimeEntity(StarlinkEntity, TimeEntity):
"""A TimeEntity for Starlink devices. Handles creating unique IDs."""
entity_description: StarlinkTimeEntityDescription
@property
def native_value(self) -> time | None:
"""Return the value reported by the time."""
return self.entity_description.value_fn(
self.coordinator.data, self.coordinator.timezone
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self.entity_description.available_fn(self.coordinator.data)
async def async_set_value(self, value: time) -> None:
"""Change the time."""
return await self.entity_description.update_fn(self.coordinator, value)
def _utc_minutes_to_time(utc_minutes: int, timezone: tzinfo) -> time:
hour = math.floor(utc_minutes / 60)
minute = utc_minutes % 60
utc = datetime.now(UTC).replace(hour=hour, minute=minute, second=0, microsecond=0)
return utc.astimezone(timezone).time()
def _time_to_utc_minutes(t: time, timezone: tzinfo) -> int:
zoned_time = datetime.now(timezone).replace(
hour=t.hour, minute=t.minute, second=0, microsecond=0
)
utc_time = zoned_time.astimezone(UTC).time()
return (utc_time.hour * 60) + utc_time.minute
TIMES = [
StarlinkTimeEntityDescription(
key="sleep_start",
translation_key="sleep_start",
value_fn=lambda data, timezone: _utc_minutes_to_time(data.sleep[0], timezone),
update_fn=lambda coordinator, time: coordinator.async_set_sleep_start(
_time_to_utc_minutes(time, coordinator.timezone)
),
available_fn=lambda data: data.sleep[2],
),
StarlinkTimeEntityDescription(
key="sleep_end",
translation_key="sleep_end",
value_fn=lambda data, timezone: _utc_minutes_to_time(
data.sleep[0] + data.sleep[1], timezone
),
update_fn=lambda coordinator, time: coordinator.async_set_sleep_duration(
_time_to_utc_minutes(time, coordinator.timezone)
),
available_fn=lambda data: data.sleep[2],
),
]

1
tests/components/starlink/fixtures/sleep_data_success.json Normal file
View File

@ -0,0 +1 @@
[0, 1, false]

5
tests/components/starlink/patchers.py
View File

@ -19,6 +19,11 @@ LOCATION_DATA_SUCCESS_PATCHER = patch(
return_value=json.loads(load_fixture("location_data_success.json", "starlink")),
)
SLEEP_DATA_SUCCESS_PATCHER = patch(
"homeassistant.components.starlink.coordinator.get_sleep_config",
return_value=json.loads(load_fixture("sleep_data_success.json", "starlink")),
)
DEVICE_FOUND_PATCHER = patch(
"homeassistant.components.starlink.config_flow.get_id", return_value="some-valid-id"
)

5
tests/components/starlink/snapshots/test_diagnostics.ambr
View File

@ -52,6 +52,11 @@
None,
]),
}),
'sleep': list([
0,
1,
False,
]),
'status': dict({
'alerts': 0,
'currently_obstructed': False,

8
tests/components/starlink/test_diagnostics.py
View File

@ -6,7 +6,11 @@ from homeassistant.components.starlink.const import DOMAIN
from homeassistant.const import CONF_IP_ADDRESS
from homeassistant.core import HomeAssistant
from .patchers import LOCATION_DATA_SUCCESS_PATCHER, STATUS_DATA_SUCCESS_PATCHER
from .patchers import (
LOCATION_DATA_SUCCESS_PATCHER,
SLEEP_DATA_SUCCESS_PATCHER,
STATUS_DATA_SUCCESS_PATCHER,
)
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
@ -24,7 +28,7 @@ async def test_diagnostics(
data={CONF_IP_ADDRESS: "1.2.3.4:0000"},
)
with STATUS_DATA_SUCCESS_PATCHER, LOCATION_DATA_SUCCESS_PATCHER:
with STATUS_DATA_SUCCESS_PATCHER, LOCATION_DATA_SUCCESS_PATCHER, SLEEP_DATA_SUCCESS_PATCHER:
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)

10
tests/components/starlink/test_init.py
View File

@ -5,7 +5,11 @@ from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_IP_ADDRESS
from homeassistant.core import HomeAssistant
from .patchers import LOCATION_DATA_SUCCESS_PATCHER, STATUS_DATA_SUCCESS_PATCHER
from .patchers import (
LOCATION_DATA_SUCCESS_PATCHER,
SLEEP_DATA_SUCCESS_PATCHER,
STATUS_DATA_SUCCESS_PATCHER,
)
from tests.common import MockConfigEntry
@ -17,7 +21,7 @@ async def test_successful_entry(hass: HomeAssistant) -> None:
data={CONF_IP_ADDRESS: "1.2.3.4:0000"},
)
with STATUS_DATA_SUCCESS_PATCHER, LOCATION_DATA_SUCCESS_PATCHER:
with STATUS_DATA_SUCCESS_PATCHER, LOCATION_DATA_SUCCESS_PATCHER, SLEEP_DATA_SUCCESS_PATCHER:
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
@ -34,7 +38,7 @@ async def test_unload_entry(hass: HomeAssistant) -> None:
data={CONF_IP_ADDRESS: "1.2.3.4:0000"},
)
with STATUS_DATA_SUCCESS_PATCHER, LOCATION_DATA_SUCCESS_PATCHER:
with STATUS_DATA_SUCCESS_PATCHER, LOCATION_DATA_SUCCESS_PATCHER, SLEEP_DATA_SUCCESS_PATCHER:
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)