Add next starship launch sensor to launch_library (#64929)

Co-authored-by: Joakim Sørensen <hi@ludeeus.dev>
This commit is contained in:
Simon Hansen 2022-01-26 12:31:03 +01:00 committed by GitHub
parent bcd7390488
commit 4e808133f2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 61 additions and 23 deletions

View File

@ -1,8 +1,13 @@
"""The launch_library component.""" """The launch_library component."""
from __future__ import annotations
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import TypedDict
from pylaunches import PyLaunches, PyLaunchesException from pylaunches import PyLaunches, PyLaunchesException
from pylaunches.objects.launch import Launch
from pylaunches.objects.starship import StarshipResponse
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform from homeassistant.const import Platform
@ -17,6 +22,13 @@ _LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.SENSOR] PLATFORMS = [Platform.SENSOR]
class LaunchLibraryData(TypedDict):
"""Typed dict representation of data returned from pylaunches."""
upcoming_launches: list[Launch]
starship_events: StarshipResponse
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up this integration using UI.""" """Set up this integration using UI."""
@ -25,9 +37,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
session = async_get_clientsession(hass) session = async_get_clientsession(hass)
launches = PyLaunches(session) launches = PyLaunches(session)
async def async_update(): async def async_update() -> LaunchLibraryData:
try: try:
return await launches.upcoming_launches() return LaunchLibraryData(
upcoming_launches=await launches.upcoming_launches(),
starship_events=await launches.starship_events(),
)
except PyLaunchesException as ex: except PyLaunchesException as ex:
raise UpdateFailed(ex) from ex raise UpdateFailed(ex) from ex

View File

@ -3,12 +3,11 @@ from __future__ import annotations
from typing import Any from typing import Any
from pylaunches.objects.launch import Launch
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import LaunchLibraryData
from .const import DOMAIN from .const import DOMAIN
@ -17,8 +16,12 @@ async def async_get_config_entry_diagnostics(
entry: ConfigEntry, entry: ConfigEntry,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Return diagnostics for a config entry.""" """Return diagnostics for a config entry."""
coordinator: DataUpdateCoordinator[list[Launch]] = hass.data[DOMAIN] coordinator: DataUpdateCoordinator[LaunchLibraryData] = hass.data[DOMAIN]
next_launch = coordinator.data[0] if coordinator.data else None if coordinator.data is None:
return {}
next_launch = coordinator.data["upcoming_launches"][0]
starship_launch = coordinator.data["starship_events"].upcoming.launches[0]
return { return {
"next_launch": next_launch.raw_data_contents if next_launch else None, "next_launch": next_launch.raw_data_contents,
"starship": starship_launch.raw_data_contents,
} }

View File

@ -28,6 +28,7 @@ from homeassistant.helpers.update_coordinator import (
) )
from homeassistant.util.dt import parse_datetime from homeassistant.util.dt import parse_datetime
from . import LaunchLibraryData
from .const import DOMAIN from .const import DOMAIN
DEFAULT_NEXT_LAUNCH_NAME = "Next launch" DEFAULT_NEXT_LAUNCH_NAME = "Next launch"
@ -41,7 +42,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
@dataclass @dataclass
class NextLaunchSensorEntityDescriptionMixin: class LaunchLibrarySensorEntityDescriptionMixin:
"""Mixin for required keys.""" """Mixin for required keys."""
value_fn: Callable[[Launch], datetime | int | str | None] value_fn: Callable[[Launch], datetime | int | str | None]
@ -49,14 +50,14 @@ class NextLaunchSensorEntityDescriptionMixin:
@dataclass @dataclass
class NextLaunchSensorEntityDescription( class LaunchLibrarySensorEntityDescription(
SensorEntityDescription, NextLaunchSensorEntityDescriptionMixin SensorEntityDescription, LaunchLibrarySensorEntityDescriptionMixin
): ):
"""Describes a Next Launch sensor entity.""" """Describes a Next Launch sensor entity."""
SENSOR_DESCRIPTIONS: tuple[NextLaunchSensorEntityDescription, ...] = ( SENSOR_DESCRIPTIONS: tuple[LaunchLibrarySensorEntityDescription, ...] = (
NextLaunchSensorEntityDescription( LaunchLibrarySensorEntityDescription(
key="next_launch", key="next_launch",
icon="mdi:rocket-launch", icon="mdi:rocket-launch",
name="Next launch", name="Next launch",
@ -68,7 +69,7 @@ SENSOR_DESCRIPTIONS: tuple[NextLaunchSensorEntityDescription, ...] = (
"provider_country_code": nl.pad.location.country_code, "provider_country_code": nl.pad.location.country_code,
}, },
), ),
NextLaunchSensorEntityDescription( LaunchLibrarySensorEntityDescription(
key="launch_time", key="launch_time",
icon="mdi:clock-outline", icon="mdi:clock-outline",
name="Launch time", name="Launch time",
@ -80,7 +81,7 @@ SENSOR_DESCRIPTIONS: tuple[NextLaunchSensorEntityDescription, ...] = (
"stream_live": nl.webcast_live, "stream_live": nl.webcast_live,
}, },
), ),
NextLaunchSensorEntityDescription( LaunchLibrarySensorEntityDescription(
key="launch_probability", key="launch_probability",
icon="mdi:dice-multiple", icon="mdi:dice-multiple",
name="Launch Probability", name="Launch Probability",
@ -88,14 +89,14 @@ SENSOR_DESCRIPTIONS: tuple[NextLaunchSensorEntityDescription, ...] = (
value_fn=lambda nl: None if nl.probability == -1 else nl.probability, value_fn=lambda nl: None if nl.probability == -1 else nl.probability,
attributes_fn=lambda nl: None, attributes_fn=lambda nl: None,
), ),
NextLaunchSensorEntityDescription( LaunchLibrarySensorEntityDescription(
key="launch_status", key="launch_status",
icon="mdi:rocket-launch", icon="mdi:rocket-launch",
name="Launch status", name="Launch status",
value_fn=lambda nl: nl.status.name, value_fn=lambda nl: nl.status.name,
attributes_fn=lambda nl: {"reason": nl.holdreason} if nl.inhold else None, attributes_fn=lambda nl: {"reason": nl.holdreason} if nl.inhold else None,
), ),
NextLaunchSensorEntityDescription( LaunchLibrarySensorEntityDescription(
key="launch_mission", key="launch_mission",
icon="mdi:orbit", icon="mdi:orbit",
name="Launch mission", name="Launch mission",
@ -106,6 +107,19 @@ SENSOR_DESCRIPTIONS: tuple[NextLaunchSensorEntityDescription, ...] = (
"description": nl.mission.description, "description": nl.mission.description,
}, },
), ),
LaunchLibrarySensorEntityDescription(
key="starship_launch",
icon="mdi:rocket",
name="Next Starship launch",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=lambda sl: parse_datetime(sl.net),
attributes_fn=lambda sl: {
"title": sl.mission.name,
"status": sl.status.name,
"target_orbit": sl.mission.orbit.name,
"description": sl.mission.description,
},
),
) )
@ -138,10 +152,10 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up the sensor platform.""" """Set up the sensor platform."""
name = entry.data.get(CONF_NAME, DEFAULT_NEXT_LAUNCH_NAME) name = entry.data.get(CONF_NAME, DEFAULT_NEXT_LAUNCH_NAME)
coordinator = hass.data[DOMAIN] coordinator: DataUpdateCoordinator[LaunchLibraryData] = hass.data[DOMAIN]
async_add_entities( async_add_entities(
NextLaunchSensor( LaunchLibrarySensor(
coordinator=coordinator, coordinator=coordinator,
entry_id=entry.entry_id, entry_id=entry.entry_id,
description=description, description=description,
@ -151,18 +165,19 @@ async def async_setup_entry(
) )
class NextLaunchSensor(CoordinatorEntity, SensorEntity): class LaunchLibrarySensor(CoordinatorEntity, SensorEntity):
"""Representation of the next launch sensors.""" """Representation of the next launch sensors."""
_attr_attribution = "Data provided by Launch Library." _attr_attribution = "Data provided by Launch Library."
_next_launch: Launch | None = None _next_launch: Launch | None = None
entity_description: NextLaunchSensorEntityDescription entity_description: LaunchLibrarySensorEntityDescription
coordinator: DataUpdateCoordinator[LaunchLibraryData]
def __init__( def __init__(
self, self,
coordinator: DataUpdateCoordinator, coordinator: DataUpdateCoordinator[LaunchLibraryData],
entry_id: str, entry_id: str,
description: NextLaunchSensorEntityDescription, description: LaunchLibrarySensorEntityDescription,
name: str | None = None, name: str | None = None,
) -> None: ) -> None:
"""Initialize a Launch Library sensor.""" """Initialize a Launch Library sensor."""
@ -194,7 +209,12 @@ class NextLaunchSensor(CoordinatorEntity, SensorEntity):
@callback @callback
def _handle_coordinator_update(self) -> None: def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator.""" """Handle updated data from the coordinator."""
self._next_launch = next((launch for launch in self.coordinator.data), None) if self.entity_description.key == "starship_launch":
launches = self.coordinator.data["starship_events"].upcoming.launches
else:
launches = self.coordinator.data["upcoming_launches"]
self._next_launch = next((launch for launch in (launches)), None)
super()._handle_coordinator_update() super()._handle_coordinator_update()
async def async_added_to_hass(self) -> None: async def async_added_to_hass(self) -> None: