starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-10-06 14:29:21 -04:00
Code Issues Packages Projects Releases Wiki Activity

indi_allsky: Add image platform for keogram and star trail images (#182970)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Hamish
2026-09-30 11:57:29 +01:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent 98e2663822
commit 82d6b1aedd
10 changed files with 448 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/indi_allsky/__init__.py
+6 -1
View File
@@ -5,7 +5,12 @@ from homeassistant.core import HomeAssistant
from .coordinator import IndiAllSkyConfigEntry, IndiAllSkyDataUpdateCoordinator
_PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.CAMERA, Platform.SENSOR]
_PLATFORMS: list[Platform] = [
Platform.BINARY_SENSOR,
Platform.CAMERA,
Platform.IMAGE,
Platform.SENSOR,
]
async def async_setup_entry(hass: HomeAssistant, entry: IndiAllSkyConfigEntry) -> bool:
homeassistant/components/indi_allsky/camera.py
+1 -1
View File
@@ -28,7 +28,7 @@ async def async_setup_entry(
class IndiAllSkyCamera(IndiAllSkyEntity, Camera):
"""Representation of an INDI Allsky camera."""
_attr_name = None
translation_key = "latest_capture"
def __init__(
self,
homeassistant/components/indi_allsky/coordinator.py
+51 -1
View File
@@ -4,7 +4,13 @@ from dataclasses import dataclass
import logging
from typing import override
from aioindiallsky import ExposureData, IndiAllSkyClient, IndiAllSkyError, SensorData
from aioindiallsky import (
ExposureData,
IndiAllSkyClient,
IndiAllSkyError,
MediaData,
SensorData,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL
@@ -25,6 +31,8 @@ class IndiAllSkyData:
"""Data model for INDI Allsky coordinator data."""
exposure: ExposureData | None = None
latest_keogram: MediaData | None = None
latest_startrail: MediaData | None = None
sensor: SensorData | None = None
@@ -43,6 +51,8 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
session=async_get_clientsession(hass),
)
self.latest_exposure: ExposureData | None = None
self.latest_keogram: MediaData | None = None
self.latest_startrail: MediaData | None = None
self.latest_sensor: SensorData | None = None
entry.async_on_unload(
@@ -50,6 +60,16 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
"exposure_complete", self._handle_exposure_complete
)
)
entry.async_on_unload(
self.client.register_callback(
"keogram_complete", self._handle_keogram_complete
)
)
entry.async_on_unload(
self.client.register_callback(
"startrail_complete", self._handle_startrail_complete
)
)
entry.async_on_unload(
self.client.register_callback("sensor_update", self._handle_sensor_update)
)
@@ -69,6 +89,32 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
self.async_set_updated_data(
IndiAllSkyData(
exposure=exposure,
latest_keogram=self.latest_keogram,
latest_startrail=self.latest_startrail,
sensor=self.latest_sensor,
)
)
def _handle_keogram_complete(self, media: MediaData) -> None:
"""Handle new keogram_complete event from WebSocket stream."""
self.latest_keogram = media
self.async_set_updated_data(
IndiAllSkyData(
exposure=self.latest_exposure,
latest_keogram=media,
latest_startrail=self.latest_startrail,
sensor=self.latest_sensor,
)
)
def _handle_startrail_complete(self, media: MediaData) -> None:
"""Handle new startrail_complete event from WebSocket stream."""
self.latest_startrail = media
self.async_set_updated_data(
IndiAllSkyData(
exposure=self.latest_exposure,
latest_keogram=self.latest_keogram,
latest_startrail=media,
sensor=self.latest_sensor,
)
)
@@ -137,6 +183,8 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
self.async_set_updated_data(
IndiAllSkyData(
exposure=self.latest_exposure,
latest_keogram=self.latest_keogram,
latest_startrail=self.latest_startrail,
sensor=self.latest_sensor,
)
)
@@ -158,5 +206,7 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
return IndiAllSkyData(
exposure=self.latest_exposure,
latest_keogram=self.latest_keogram,
latest_startrail=self.latest_startrail,
sensor=self.latest_sensor,
)
homeassistant/components/indi_allsky/image.py
+116
View File
@@ -0,0 +1,116 @@
"""Support for INDI Allsky image entities."""
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from typing import override
from aioindiallsky import IndiAllSkyError, MediaData
from homeassistant.components.image import (
ImageEntity,
ImageEntityDescription,
infer_image_type,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .coordinator import (
IndiAllSkyConfigEntry,
IndiAllSkyData,
IndiAllSkyDataUpdateCoordinator,
)
from .entity import IndiAllSkyEntity
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class IndiAllSkyImageEntityDescription(ImageEntityDescription):
"""Class describing INDI Allsky image entities."""
media_fn: Callable[[IndiAllSkyData], MediaData | None]
fallback_filename: str
IMAGE_DESCRIPTIONS: tuple[IndiAllSkyImageEntityDescription, ...] = (
IndiAllSkyImageEntityDescription(
key="latest_keogram",
translation_key="latest_keogram",
media_fn=lambda data: data.latest_keogram,
fallback_filename="latestkeogram",
),
IndiAllSkyImageEntityDescription(
key="latest_startrail",
translation_key="latest_startrail",
media_fn=lambda data: data.latest_startrail,
fallback_filename="lateststartrail",
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: IndiAllSkyConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up INDI Allsky image entities based on a config entry."""
coordinator = entry.runtime_data
async_add_entities(
IndiAllSkyImageEntity(hass, coordinator, entry, description)
for description in IMAGE_DESCRIPTIONS
)
class IndiAllSkyImageEntity(IndiAllSkyEntity, ImageEntity):
"""Representation of an INDI Allsky image entity."""
entity_description: IndiAllSkyImageEntityDescription
def __init__(
self,
hass: HomeAssistant,
coordinator: IndiAllSkyDataUpdateCoordinator,
entry: IndiAllSkyConfigEntry,
description: IndiAllSkyImageEntityDescription,
) -> None:
"""Initialize the image entity."""
super().__init__(coordinator, entry)
ImageEntity.__init__(self, hass)
self.entity_description = description
self._attr_unique_id = f"{entry.entry_id}_{description.key}"
self._last_fetched: datetime | None = None
@property
@override
def image_last_updated(self) -> datetime | None:
"""Return the timestamp when the image was last updated."""
media = self.entity_description.media_fn(self.coordinator.data)
if media and media.day_date:
if dt := dt_util.parse_datetime(media.day_date):
if dt.tzinfo is None:
return dt.replace(tzinfo=dt_util.UTC)
return dt_util.as_utc(dt)
return self._last_fetched
@override
async def async_image(self) -> bytes | None:
"""Return bytes of the image."""
media = self.entity_description.media_fn(self.coordinator.data)
filename = (
media.filename
if (media and media.filename)
else self.entity_description.fallback_filename
)
try:
image_bytes = await self.coordinator.client.fetch_image(filename)
except IndiAllSkyError:
return None
else:
if content_type := infer_image_type(image_bytes):
self._attr_content_type = content_type
if media is None and self._last_fetched is None:
self._last_fetched = dt_util.utcnow()
self.async_write_ha_state()
return image_bytes
homeassistant/components/indi_allsky/strings.json
+13
View File
@@ -32,6 +32,19 @@
"name": "Night"
}
},
"camera": {
"latest_capture": {
"name": "Latest capture"
}
},
"image": {
"latest_keogram": {
"name": "Latest keogram"
},
"latest_startrail": {
"name": "Latest star trail"
}
},
"sensor": {
"ambient_temperature": {
"name": "Ambient temperature"
tests/components/indi_allsky/snapshots/test_camera.ambr
+9 -9
View File
@@ -1,5 +1,5 @@
# serializer version: 1
# name: test_camera_setup_and_states[camera.indi_allsky-entry]
# name: test_camera_setup_and_states[camera.indi_allsky_latest_capture-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
@@ -13,7 +13,7 @@
'disabled_by': None,
'domain': 'camera',
'entity_category': None,
'entity_id': 'camera.indi_allsky',
'entity_id': 'camera.indi_allsky_latest_capture',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
@@ -21,31 +21,31 @@
'labels': set({
}),
'name': None,
'object_id_base': None,
'object_id_base': 'Latest capture',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'original_name': 'Latest capture',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'translation_key': 'latest_capture',
'unique_id': '1234567890abcdef1234567890abcdef',
'unit_of_measurement': None,
})
# ---
# name: test_camera_setup_and_states[camera.indi_allsky-state]
# name: test_camera_setup_and_states[camera.indi_allsky_latest_capture-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<CameraEntityStateAttribute.ACCESS_TOKEN: 'access_token'>: '1caab5c3b3',
<EntityStateAttribute.ENTITY_PICTURE: 'entity_picture'>: '/api/camera_proxy/camera.indi_allsky?token=1caab5c3b3',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky',
<EntityStateAttribute.ENTITY_PICTURE: 'entity_picture'>: '/api/camera_proxy/camera.indi_allsky_latest_capture?token=1caab5c3b3',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Latest capture',
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <CameraEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'camera.indi_allsky',
'entity_id': 'camera.indi_allsky_latest_capture',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
tests/components/indi_allsky/snapshots/test_image.ambr
+105
View File
@@ -0,0 +1,105 @@
# serializer version: 1
# name: test_image_setup_and_states[image.indi_allsky_latest_keogram-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'image',
'entity_category': None,
'entity_id': 'image.indi_allsky_latest_keogram',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Latest keogram',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Latest keogram',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'latest_keogram',
'unique_id': '1234567890abcdef1234567890abcdef_latest_keogram',
'unit_of_measurement': None,
})
# ---
# name: test_image_setup_and_states[image.indi_allsky_latest_keogram-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<ImageEntityStateAttribute.ACCESS_TOKEN: 'access_token'>: '1caab5c3b3',
<EntityStateAttribute.ENTITY_PICTURE: 'entity_picture'>: '/api/image_proxy/image.indi_allsky_latest_keogram?token=1caab5c3b3',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Latest keogram',
}),
'context': <ANY>,
'entity_id': 'image.indi_allsky_latest_keogram',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_image_setup_and_states[image.indi_allsky_latest_star_trail-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'image',
'entity_category': None,
'entity_id': 'image.indi_allsky_latest_star_trail',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Latest star trail',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Latest star trail',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'latest_startrail',
'unique_id': '1234567890abcdef1234567890abcdef_latest_startrail',
'unit_of_measurement': None,
})
# ---
# name: test_image_setup_and_states[image.indi_allsky_latest_star_trail-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<ImageEntityStateAttribute.ACCESS_TOKEN: 'access_token'>: '1caab5c3b3',
<EntityStateAttribute.ENTITY_PICTURE: 'entity_picture'>: '/api/image_proxy/image.indi_allsky_latest_star_trail?token=1caab5c3b3',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Latest star trail',
}),
'context': <ANY>,
'entity_id': 'image.indi_allsky_latest_star_trail',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
tests/components/indi_allsky/test_camera.py
+2 -2
View File
@@ -50,7 +50,7 @@ async def test_camera_image_and_update(
mock_indi_allsky_client.fetch_image.return_value = image_bytes
await setup_integration(hass, mock_config_entry)
image = await async_get_image(hass, "camera.indi_allsky")
image = await async_get_image(hass, "camera.indi_allsky_latest_capture")
assert image.content == image_bytes
assert image.content_type == expected_content_type
@@ -66,4 +66,4 @@ async def test_camera_image_fetch_failure(
mock_indi_allsky_client.fetch_image.side_effect = IndiAllSkyError("Fetch error")
with pytest.raises(HomeAssistantError, match="Unable to get image"):
await async_get_image(hass, "camera.indi_allsky")
await async_get_image(hass, "camera.indi_allsky_latest_capture")
tests/components/indi_allsky/test_image.py
+144
View File
@@ -0,0 +1,144 @@
"""Tests for the INDI Allsky image platform."""
from unittest.mock import AsyncMock, patch
from aioindiallsky import IndiAllSkyError, MediaData
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components import image
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
@pytest.fixture
def mock_keogram_data() -> MediaData:
"""Fixture to provide sample keogram MediaData."""
return MediaData.from_dict(
"keogram",
{
"filename": "keogram_20260813.jpg",
"dayDate": "2026-08-13 22:53:41",
"night": True,
"camera_id": 1,
"id": 1,
},
)
@pytest.fixture
def mock_startrail_data() -> MediaData:
"""Fixture to provide sample startrail MediaData."""
return MediaData.from_dict(
"startrail",
{
"filename": "startrail_20260813.jpg",
"dayDate": "2026-08-13 22:53:41",
"night": True,
"camera_id": 1,
"id": 2,
},
)
@pytest.mark.usefixtures("mock_indi_allsky_client")
async def test_image_setup_and_states(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
) -> None:
"""Test standard successful setup and image entity snapshots."""
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.IMAGE]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(
hass, entity_registry, snapshot, mock_config_entry.entry_id
)
async def test_image_events_and_fetching(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_keogram_data: MediaData,
mock_startrail_data: MediaData,
) -> None:
"""Test image entities update state and return image bytes on media events."""
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.IMAGE]):
await setup_integration(hass, mock_config_entry)
for callback in mock_indi_allsky_client.callbacks.get("keogram_complete", []):
callback(mock_keogram_data)
for callback in mock_indi_allsky_client.callbacks.get("startrail_complete", []):
callback(mock_startrail_data)
await hass.async_block_till_done()
state = hass.states.get("image.indi_allsky_latest_keogram")
assert state is not None
assert state.state == "2026-08-13T22:53:41+00:00"
mock_indi_allsky_client.fetch_image.return_value = b"\xff\xd8\xff\xe0keogram_bytes"
img = await image.async_get_image(hass, "image.indi_allsky_latest_keogram")
assert img.content == b"\xff\xd8\xff\xe0keogram_bytes"
mock_indi_allsky_client.fetch_image.assert_called_with("keogram_20260813.jpg")
state = hass.states.get("image.indi_allsky_latest_star_trail")
assert state is not None
assert state.state == "2026-08-13T22:53:41+00:00"
mock_indi_allsky_client.fetch_image.return_value = (
b"\xff\xd8\xff\xe0startrail_bytes"
)
img = await image.async_get_image(hass, "image.indi_allsky_latest_star_trail")
assert img.content == b"\xff\xd8\xff\xe0startrail_bytes"
mock_indi_allsky_client.fetch_image.assert_called_with("startrail_20260813.jpg")
async def test_image_fetch_error(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_keogram_data: MediaData,
) -> None:
"""Test handling of image fetch errors."""
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.IMAGE]):
await setup_integration(hass, mock_config_entry)
for callback in mock_indi_allsky_client.callbacks.get("keogram_complete", []):
callback(mock_keogram_data)
await hass.async_block_till_done()
mock_indi_allsky_client.fetch_image.side_effect = IndiAllSkyError("HTTP Error")
with pytest.raises(HomeAssistantError):
await image.async_get_image(hass, "image.indi_allsky_latest_keogram")
async def test_image_fallback_fetching_before_events(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test fetching fallback alias images before any media event arrives."""
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.IMAGE]):
await setup_integration(hass, mock_config_entry)
mock_indi_allsky_client.fetch_image.return_value = (
b"\xff\xd8\xff\xe0fallback_keogram"
)
img = await image.async_get_image(hass, "image.indi_allsky_latest_keogram")
assert img.content == b"\xff\xd8\xff\xe0fallback_keogram"
mock_indi_allsky_client.fetch_image.assert_called_with("latestkeogram")
mock_indi_allsky_client.fetch_image.return_value = (
b"\xff\xd8\xff\xe0fallback_startrail"
)
img = await image.async_get_image(hass, "image.indi_allsky_latest_star_trail")
assert img.content == b"\xff\xd8\xff\xe0fallback_startrail"
mock_indi_allsky_client.fetch_image.assert_called_with("lateststartrail")
tests/components/indi_allsky/test_init.py
+1 -2
View File
@@ -21,10 +21,9 @@ async def test_setup_and_unload_entry(
"""Test successful setup and unload of entry."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_indi_allsky_client.listen.assert_called_once_with(auto_reconnect=True)
await hass.config_entries.async_unload(mock_config_entry.entry_id)
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED