Max Gramser
306636434b
Live picture (issue #4): a camera or image tile on a Guition takes Display -> Live picture with a pace of 15 or 30 s. The camera tiles of a page share one download: the screen asks ESP Screens for them together (esphome.screen_camera with tiles, size and bg) and the app serves one BMP, a strip of squares with the corners rounded over each tile's own colour; every tile draws its square out of it with LVGL's image offset (LV_IMAGE_ALIGN_TOP_LEFT: the default centres a taller source). A third online_image on the Guition holds the strip; it loads after the alert's picture, a cover and the camera full screen, never under a card, in standby or under a finger. The strip is served whole every time: ESPHome's http_request logs a 304 as a failed request. Firmware 0.2.77. Alert button: esp_screens_show_alert takes action and data; the app performs the action once when the button's report (esphome.screen_alert, ok) comes from any screen. show_alert keeps its seven fields, so every firmware from 0.2.31 has it. Media card and full-page media tile: a title or artist line that is too wide rolls by (LV_LABEL_LONG_SCROLL_CIRCULAR) instead of ending in dots. Full-page tiles are white or their own pastel like every other tile; the tint in the state colour of firmware 0.2.62 to 0.2.76 is gone. Host harness (27 checks) and bench Guition + local add-on on the Yellow: docs/TEST_RESULTS_0291.md. CYD firmware 1,633,728 B, 89.0 %. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
273 lines
16 KiB
Python
273 lines
16 KiB
Python
"""The media card (app 0.2.77, firmware 0.2.64): what the app pre-computes for it, and the album cover the app serves
|
|
like a camera image (camera_feed), at the size the card asks for, with the corners rounded over the colour behind them."""
|
|
import asyncio
|
|
import importlib.util
|
|
import io
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / 'screen_manager/app'))
|
|
import camera_feed # noqa: E402
|
|
from core import COVER_MIN_FIRMWARE, extras, media_extras, state_message # noqa: E402
|
|
|
|
HAS_PIL = importlib.util.find_spec('PIL') is not None
|
|
HAS_AIOHTTP = importlib.util.find_spec('aiohttp') is not None
|
|
if HAS_AIOHTTP:
|
|
from server import Manager
|
|
|
|
TILES = (ROOT / 'components/smart_display/runtime_tiles.h').read_text()
|
|
MODEL = (ROOT / 'components/smart_display/runtime_model.h').read_text()
|
|
CARD = (ROOT / 'components/smart_display/media_card.h').read_text()
|
|
SONOS = {'friendly_name': 'Office', 'media_title': 'Woman at the Loom', 'media_artist': 'KATZROAR', 'media_album_name': 'Invocation',
|
|
'media_duration': 270, 'media_position': 0, 'media_position_updated_at': '2026-09-18T10:09:49.068667+00:00',
|
|
'volume_level': 0.14, 'is_volume_muted': False, 'supported_features': 8321599,
|
|
'entity_picture': '/api/media_player_proxy/media_player.office?token=5674b25e&cache=556b49fd9c3e8ccd'}
|
|
|
|
|
|
def picture(size=(640, 640), mode='RGB', fmt='JPEG'):
|
|
from PIL import Image
|
|
out = io.BytesIO()
|
|
Image.new(mode, size, (200, 30, 90) if mode == 'RGB' else None).save(out, fmt)
|
|
return out.getvalue()
|
|
|
|
|
|
class Extras(unittest.TestCase):
|
|
def test_the_card_gets_artist_album_length_position_and_a_picture_mark(self):
|
|
extra = media_extras(SONOS)
|
|
self.assertEqual(extra['artist'], 'KATZROAR')
|
|
self.assertEqual(extra['album'], 'Invocation')
|
|
self.assertEqual(extra['dur'], 270)
|
|
self.assertEqual(extra['pos'], 0)
|
|
self.assertEqual(extra['at'], 1789726189) # 2026-09-18T10:09:49Z
|
|
self.assertTrue(re.fullmatch(r'[0-9a-f]{10}', extra['pic']))
|
|
# The mark follows the picture and nothing else: a new cache token is a new picture.
|
|
other = media_extras({**SONOS, 'entity_picture': SONOS['entity_picture'].replace('556b49fd9c3e8ccd', '1')})
|
|
self.assertNotEqual(other['pic'], extra['pic'])
|
|
self.assertEqual(media_extras({**SONOS, 'volume_level': 0.5})['pic'], extra['pic'])
|
|
|
|
def test_a_player_without_a_track_gets_no_extras(self):
|
|
self.assertIsNone(media_extras({'friendly_name': 'Bedroom', 'supported_features': 8321599}))
|
|
self.assertIsNone(media_extras({'media_artist': None, 'media_duration': None, 'media_position_updated_at': None}))
|
|
# A stream: a title, no length, no picture. The card shows an empty bar and no times.
|
|
self.assertIsNone(media_extras({'media_title': 'NPO Radio 2', 'media_content_type': 'channel'}))
|
|
|
|
def test_odd_values_stay_away(self):
|
|
extra = media_extras({**SONOS, 'media_duration': -5, 'media_position': float('nan'), 'media_position_updated_at': 'yesterday',
|
|
'media_artist': ' ', 'entity_picture': 42})
|
|
self.assertNotIn('dur', extra)
|
|
self.assertNotIn('pos', extra)
|
|
self.assertNotIn('at', extra)
|
|
self.assertNotIn('artist', extra)
|
|
self.assertNotIn('pic', extra)
|
|
self.assertEqual(extra['album'], 'Invocation')
|
|
# Long names are cut like every text on a screen; a picture the state names as local counts too.
|
|
extra = media_extras({'media_artist': 'x' * 200, 'entity_picture_local': '/api/media_player_proxy/a?token=b'})
|
|
self.assertEqual(len(extra['artist']), 80)
|
|
self.assertIn('pic', extra)
|
|
|
|
def test_the_state_message_carries_them_and_stays_small(self):
|
|
states = {'media_player.office': {'state': 'playing', 'attributes': SONOS}}
|
|
tile = {'entity': 'media_player.office', 'name': ''}
|
|
message = state_message(0, tile, states, extras(tile, states))
|
|
self.assertEqual(message['x']['artist'], 'KATZROAR')
|
|
self.assertEqual(message['a']['media_title'], 'Woman at the Loom')
|
|
self.assertLess(len(str(message)), 700)
|
|
|
|
|
|
class Firmware(unittest.TestCase):
|
|
def test_the_firmware_reads_what_the_app_sends(self):
|
|
for key in ('extra["artist"]', 'extra["album"]', 'extra["pic"]', 'extra["dur"]', 'extra["pos"]', 'extra["at"]'):
|
|
self.assertIn(key, TILES, key)
|
|
for field in ('media_artist', 'media_album', 'media_picture', 'media_duration', 'media_position', 'media_position_at'):
|
|
self.assertIn(field, MODEL, field)
|
|
self.assertIn(field, MODEL[MODEL.index('bool empty() const'):], f'{field} counted in Extra::empty')
|
|
|
|
def test_the_card_asks_for_a_cover_with_size_and_background(self):
|
|
self.assertIn('"size", "bg"', TILES)
|
|
self.assertIn('view != "cover"', TILES)
|
|
self.assertIn('if (view == "cover")', TILES)
|
|
# The keys of the card and of the tile over the whole page share one action table.
|
|
self.assertIn('inline void media_action(Tile &t,int cmd)', TILES)
|
|
self.assertIn('media_player.volume_mute', TILES)
|
|
# The cover goes before the camera loads into the same buffer, and comes back after.
|
|
self.assertIn('cover_release();\n if (camera_release_due) camera_release();', TILES)
|
|
|
|
def test_the_layout_has_no_colour_of_its_own(self):
|
|
self.assertIsNone(re.search(r'0x[0-9A-Fa-f]{6}', CARD))
|
|
self.assertNotIn('lv_', CARD)
|
|
|
|
|
|
class Covers(unittest.TestCase):
|
|
def test_a_media_player_may_ask_for_a_cover_and_a_camera_may_not(self):
|
|
self.assertTrue(camera_feed.cover_supported('media_player.office'))
|
|
self.assertFalse(camera_feed.cover_supported('camera.front_door'))
|
|
self.assertFalse(camera_feed.cover_supported('media_player.Office'))
|
|
self.assertFalse(camera_feed.supported('media_player.office'))
|
|
self.assertEqual(camera_feed.COVER_MIN_FIRMWARE, COVER_MIN_FIRMWARE)
|
|
self.assertTrue(camera_feed.can_show_cover({'board': 'guition', 'firmware': '0.2.64'}))
|
|
self.assertFalse(camera_feed.can_show_cover({'board': 'guition', 'firmware': '0.2.63'}))
|
|
self.assertFalse(camera_feed.can_show_cover({'board': 'cyd', 'firmware': '0.2.64'}))
|
|
|
|
def test_the_request_names_a_size_and_a_background(self):
|
|
self.assertEqual(camera_feed.cover_request({'size': '160', 'bg': 'E7E7E7'}), (160, 0xE7E7E7))
|
|
self.assertEqual(camera_feed.cover_request({'size': 200, 'bg': '1a1a1a'}), (200, 0x1A1A1A))
|
|
for bad in ({'size': '9999', 'bg': 'FFFFFF'}, {'size': '10', 'bg': 'FFFFFF'}, {'size': '160', 'bg': 'white'},
|
|
{'size': '160'}, {'bg': 'FFFFFF'}, {'size': 'x', 'bg': 'FFFFFF'}):
|
|
self.assertIsNone(camera_feed.cover_request(bad), bad)
|
|
|
|
@unittest.skipUnless(HAS_PIL, 'Pillow')
|
|
def test_the_cover_is_square_at_the_size_asked_with_rounded_corners_over_the_background(self):
|
|
from PIL import Image
|
|
out = camera_feed.encode_cover(picture((640, 640), fmt='PNG'), 160, 0x1A1A1A)
|
|
with Image.open(io.BytesIO(out)) as image:
|
|
self.assertEqual((image.format, image.size, image.mode), ('BMP', (160, 160), 'RGB'))
|
|
self.assertEqual(image.getpixel((0, 0)), (26, 26, 26)) # the corner: the background
|
|
self.assertEqual(image.getpixel((80, 80)), (200, 30, 90)) # the middle: the picture
|
|
self.assertEqual(image.getpixel((0, 80)), (200, 30, 90)) # the edge between the corners: the picture
|
|
# A wide picture is cut to its middle; a PNG with transparency lands on the background.
|
|
with Image.open(io.BytesIO(camera_feed.encode_cover(picture((900, 300)), 48, 0xFFFFFF))) as image:
|
|
self.assertEqual(image.size, (48, 48))
|
|
with Image.open(io.BytesIO(camera_feed.encode_cover(picture((100, 100), 'RGBA', 'PNG'), 48, 0x00FF00))) as image:
|
|
self.assertEqual(image.size, (48, 48))
|
|
|
|
@unittest.skipUnless(HAS_PIL, 'Pillow')
|
|
def test_a_cover_is_fetched_when_the_picture_changes_and_served_at_a_link(self):
|
|
pictures = {'media_player.office': '/api/media_player_proxy/media_player.office?token=a&cache=1'}
|
|
fetched = []
|
|
|
|
async def fetch_cover(entity):
|
|
fetched.append(pictures[entity])
|
|
return picture((640, 640)) if 'cache=1' in pictures[entity] else picture((300, 300))
|
|
|
|
async def scenario():
|
|
feed = camera_feed.CameraFeed(lambda entity: None, fetch_cover=fetch_cover, picture=lambda entity: pictures.get(entity, ''))
|
|
# No picture: no cover, nothing fetched.
|
|
self.assertIsNone(await feed.cover('media_player.bedroom', 160, 0))
|
|
self.assertEqual(fetched, [])
|
|
# The first load fetches; the next load of the same picture serves what it has.
|
|
tag, image = await feed.cover('media_player.office', 160, 0xFFFFFF)
|
|
self.assertEqual(len(fetched), 1)
|
|
self.assertTrue(tag.endswith('-c160-FFFFFF"'))
|
|
again = await feed.cover('media_player.office', 160, 0xFFFFFF)
|
|
self.assertEqual(len(fetched), 1)
|
|
self.assertEqual(again, (tag, image))
|
|
# Another size or background: the same picture, sized again.
|
|
other_tag, other = await feed.cover('media_player.office', 192, 0x1A1A1A)
|
|
self.assertEqual(len(fetched), 1)
|
|
self.assertNotEqual(other_tag, tag)
|
|
self.assertNotEqual(other, image)
|
|
# A new picture in Home Assistant: fetched again on the next load.
|
|
pictures['media_player.office'] = '/api/media_player_proxy/media_player.office?token=a&cache=2'
|
|
new_tag, _ = await feed.cover('media_player.office', 160, 0xFFFFFF)
|
|
self.assertEqual(len(fetched), 2)
|
|
self.assertNotEqual(new_tag, tag)
|
|
# The link: a screen loads it, and the link answers with the cover of this moment.
|
|
token = feed.link('media_player.office', (160, 160), cover=(160, 0xFFFFFF))
|
|
status, body, etag = await feed.serve(token)
|
|
self.assertEqual((status, etag), (200, new_tag))
|
|
self.assertEqual(body[:2], b'BM')
|
|
self.assertEqual((await feed.serve(token, etag))[0], 304)
|
|
# A player whose picture went away: no cover.
|
|
pictures['media_player.office'] = ''
|
|
self.assertIsNone(await feed.cover('media_player.office', 160, 0xFFFFFF))
|
|
self.assertEqual((await feed.serve(token))[0], 503)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
@unittest.skipUnless(HAS_PIL, 'Pillow')
|
|
def test_a_failing_fetch_gives_no_cover(self):
|
|
async def failing(entity):
|
|
raise ConnectionError('gone')
|
|
|
|
async def scenario():
|
|
feed = camera_feed.CameraFeed(lambda entity: None, fetch_cover=failing, picture=lambda entity: '/api/x')
|
|
self.assertIsNone(await feed.cover('media_player.office', 160, 0))
|
|
self.assertEqual(feed.watches['media_player.office'].failures, 1)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@unittest.skipUnless(HAS_AIOHTTP and HAS_PIL, 'aiohttp and Pillow')
|
|
class Answer(unittest.TestCase):
|
|
"""The app's answer to a screen's request, through Manager.answer_camera with a Home Assistant stand-in."""
|
|
|
|
def setUp(self):
|
|
# base_url keeps the last address for ten minutes; another test's Home Assistant may have left one behind.
|
|
camera_feed.base_url.__defaults__[0].clear()
|
|
|
|
class HA:
|
|
online = True
|
|
|
|
def __init__(self, firmware='0.2.64', guition=True):
|
|
self.registry = [{'entity_id': 'text.hall_tile_settings', 'platform': 'esphome', 'original_name': 'Tile settings', 'device_id': 'd1'},
|
|
{'entity_id': 'sensor.hall_node', 'platform': 'esphome', 'original_name': 'Device name', 'device_id': 'd1'},
|
|
{'entity_id': 'sensor.hall_fw', 'platform': 'esphome', 'original_name': 'Screen firmware', 'device_id': 'd1'}]
|
|
if guition:
|
|
self.registry.append({'entity_id': 'select.hall_type', 'platform': 'esphome', 'original_name': 'Guition screen type', 'device_id': 'd1'})
|
|
self.devices, self.areas = [{'id': 'd1', 'name': 'Hall'}], []
|
|
self.states = {'text.hall_tile_settings': {'state': 'Synced'}, 'sensor.hall_node': {'state': 'hall'}, 'sensor.hall_fw': {'state': firmware},
|
|
'media_player.office': {'state': 'playing', 'attributes': SONOS},
|
|
'media_player.radio': {'state': 'playing', 'attributes': {'media_title': 'Radio 2', 'supported_features': 8321599}}}
|
|
self.sent, self.responses, self.time_zone, self.changed = [], set(), None, asyncio.Event()
|
|
|
|
async def send(self, inbox, message, action=None, respond=False):
|
|
self.sent.append((inbox, message))
|
|
|
|
async def request(self, kind, **data):
|
|
return {'adapters': [{'default': True, 'ipv4': [{'address': '192.168.1.5'}]}]} if kind == 'network' else {}
|
|
|
|
def media_picture(self, entity):
|
|
attrs = self.states.get(entity, {}).get('attributes', {})
|
|
return attrs.get('entity_picture') or ''
|
|
|
|
async def media_image(self, entity):
|
|
return picture()
|
|
|
|
def manager(self, ha):
|
|
import tempfile
|
|
manager = Manager(ha, Path(tempfile.mkdtemp()) / 'screens.json')
|
|
manager.save('text.hall_tile_settings', {'title': 'Hall', 'tiles': [{'entity': 'media_player.office', 'name': ''}, {'entity': 'media_player.radio', 'name': ''}]})
|
|
return manager
|
|
|
|
def test_a_guition_card_gets_a_link_a_stream_an_empty_one_and_a_stranger_nothing(self):
|
|
ha = self.HA()
|
|
manager = self.manager(ha)
|
|
|
|
async def scenario():
|
|
await manager.answer_camera({'inbox': 'text.hall_tile_settings', 'entity': 'media_player.office', 'size': '160', 'bg': 'E7E7E7'})
|
|
self.assertEqual(len(ha.sent), 1)
|
|
message = ha.sent[0][1]
|
|
self.assertEqual((message['op'], message['t'], message['e']), ('camera', 'cover', 'media_player.office'))
|
|
self.assertTrue(message['u'].startswith('http://192.168.1.5:') and message['u'].endswith('.bmp'))
|
|
token = message['u'].rsplit('/', 1)[1][:-4]
|
|
status, body, _ = await manager.camera.serve(token)
|
|
self.assertEqual(status, 200)
|
|
from PIL import Image
|
|
with Image.open(io.BytesIO(body)) as image:
|
|
self.assertEqual(image.size, (160, 160))
|
|
self.assertEqual(image.getpixel((0, 0)), (0xE7, 0xE7, 0xE7))
|
|
# A stream without a picture: an empty link, so the card keeps its placeholder and stops asking.
|
|
await manager.answer_camera({'inbox': 'text.hall_tile_settings', 'entity': 'media_player.radio', 'size': '160', 'bg': 'E7E7E7'})
|
|
self.assertEqual(ha.sent[1][1], {'v': 1, 'op': 'camera', 't': 'cover', 'e': 'media_player.radio', 'u': ''})
|
|
# A player that is not on the screen's layout, a size out of range, a request without a size: no answer.
|
|
for request in ({'inbox': 'text.hall_tile_settings', 'entity': 'media_player.other', 'size': '160', 'bg': 'E7E7E7'},
|
|
{'inbox': 'text.hall_tile_settings', 'entity': 'media_player.office', 'size': '2000', 'bg': 'E7E7E7'},
|
|
{'inbox': 'text.hall_tile_settings', 'entity': 'media_player.office'}):
|
|
await manager.answer_camera(request)
|
|
self.assertEqual(len(ha.sent), 2)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
def test_older_firmware_and_a_cyd_get_no_cover(self):
|
|
for ha in (self.HA(firmware='0.2.63'), self.HA(guition=False)):
|
|
manager = self.manager(ha)
|
|
asyncio.run(manager.answer_camera({'inbox': 'text.hall_tile_settings', 'entity': 'media_player.office', 'size': '160', 'bg': 'E7E7E7'}))
|
|
self.assertEqual(ha.sent, [])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|