A media card like a phone's now playing, with the album cover on a Guition (0.2.77)
The card behind a media tile shows the album cover, the title, the artist and the album, a progress bar that runs while the track plays with the times at its ends, round keys for previous, play or pause and next, and a volume row with a mute key, a wide slider and the percentage. Keys the player lacks stay faded, an off player shows one power key, a stream without a length shows no bar, and the CYD shows the player's icon in the cover's place. A media tile over the whole page is the same card under the tile's head. media_card.h holds the geometry, tests/test_media_card.cpp checks every shape. The cover comes the way camera images do: the screen asks with the size and the colour behind the corners, the app fetches the picture from Home Assistant, cuts it square, rounds the corners over that colour and serves it on port 8098; the Guition loads it once and asks again only when the picture's mark in the state changes. Firmware 0.2.64. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
@@ -23,6 +23,14 @@ LOG = logging.getLogger(__name__)
|
||||
|
||||
DOMAINS = ('camera', 'image')
|
||||
MIN_FIRMWARE = (0, 2, 57)
|
||||
# Album covers on the media card (app 0.2.77, firmware 0.2.64): the same port serves a media player's picture, square,
|
||||
# at the size the screen asks for, with the rounded corners baked in over the colour the screen names (the card's
|
||||
# page or the tile's own colour), so the screen draws it as it is. A cover is fetched when the screen loads its link
|
||||
# and again only when Home Assistant's picture changes; the screen asks for a new link then.
|
||||
COVER_DOMAINS = ('media_player',)
|
||||
COVER_MIN_FIRMWARE = (0, 2, 64)
|
||||
COVER_SIZES = (48, 320) # the smallest and largest cover a screen may ask for, in pixels
|
||||
COVER_RADIUS_SHARE = 12 # the corner is a twelfth of the size, at least 4 px (media_card.h radius_for)
|
||||
PORT = 8098
|
||||
# A camera nobody loaded a picture of for this long is forgotten, its last snapshot with it.
|
||||
WATCH_SECONDS = 30
|
||||
@@ -60,6 +68,30 @@ def can_show(screen):
|
||||
return bool(screen) and screen.get('board') in BOXES and version(screen.get('firmware')) >= MIN_FIRMWARE
|
||||
|
||||
|
||||
def cover_supported(entity):
|
||||
"""True for a media player entity id: its cover can be served."""
|
||||
return (isinstance(entity, str) and len(entity) <= 120 and re.fullmatch(r'[a-z0-9_]+\.[a-z0-9_]+', entity) is not None
|
||||
and entity.split('.')[0] in COVER_DOMAINS)
|
||||
|
||||
|
||||
def can_show_cover(screen):
|
||||
"""A paired Guition with firmware that draws the media card's cover."""
|
||||
return bool(screen) and screen.get('board') in BOXES and version(screen.get('firmware')) >= COVER_MIN_FIRMWARE
|
||||
|
||||
|
||||
def cover_request(request):
|
||||
"""(size, background) a screen asked for, or None when the request is not a cover request the app can serve:
|
||||
`size` in pixels within COVER_SIZES, `bg` six hex digits of the colour behind the corners."""
|
||||
try:
|
||||
size = int(request.get('size'))
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
background = request.get('bg')
|
||||
if not COVER_SIZES[0] <= size <= COVER_SIZES[1] or not isinstance(background, str) or not re.fullmatch(r'[0-9A-Fa-f]{6}', background):
|
||||
return None
|
||||
return size, int(background, 16)
|
||||
|
||||
|
||||
def fit(size, box):
|
||||
"""The largest size with the proportions of `size` that fits `box`, at least one pixel each way."""
|
||||
width, height = size
|
||||
@@ -90,12 +122,46 @@ def encode(raw, box):
|
||||
return out.getvalue()
|
||||
|
||||
|
||||
def encode_cover(raw, size, background):
|
||||
"""A media player's picture as a square 24-bit BMP of `size` pixels with rounded corners, the corners filled with
|
||||
`background` (0xRRGGBB): what the screen draws on its card without any work of its own. A picture that is not
|
||||
square is cut to its middle."""
|
||||
from PIL import Image, ImageDraw, ImageOps
|
||||
with Image.open(io.BytesIO(raw)) as source:
|
||||
source.draft('RGB', (size, size))
|
||||
image = ImageOps.exif_transpose(source)
|
||||
if image.mode in ('RGBA', 'LA', 'P', 'PA'):
|
||||
image = image.convert('RGBA')
|
||||
ground = Image.new('RGB', image.size, tuple((background >> shift) & 0xFF for shift in (16, 8, 0)))
|
||||
ground.paste(image, mask=image.getchannel('A'))
|
||||
image = ground
|
||||
elif image.mode != 'RGB':
|
||||
image = image.convert('RGB')
|
||||
width, height = image.size
|
||||
side = min(width, height)
|
||||
if width != height:
|
||||
left, top = (width - side) // 2, (height - side) // 2
|
||||
image = image.crop((left, top, left + side, top + side))
|
||||
if image.size != (size, size):
|
||||
image = image.resize((size, size), Image.Resampling.LANCZOS)
|
||||
radius = max(4, size // COVER_RADIUS_SHARE)
|
||||
mask = Image.new('L', (size, size), 0)
|
||||
ImageDraw.Draw(mask).rounded_rectangle((0, 0, size - 1, size - 1), radius=radius, fill=255)
|
||||
ground = Image.new('RGB', (size, size), tuple((background >> shift) & 0xFF for shift in (16, 8, 0)))
|
||||
ground.paste(image, mask=mask)
|
||||
out = io.BytesIO()
|
||||
ground.save(out, 'BMP')
|
||||
return out.getvalue()
|
||||
|
||||
|
||||
class Watch:
|
||||
"""One camera: its last snapshot, the sizes made of it and the fetch on its way."""
|
||||
"""One camera or media player: its last picture, the sizes made of it and the fetch on its way. For a media
|
||||
player `picture` is the address the raw picture came from, so a changed picture is fetched again."""
|
||||
|
||||
def __init__(self, now):
|
||||
self.used = now
|
||||
self.raw, self.digest = None, ''
|
||||
self.picture, self.fetching = None, None
|
||||
self.frames = {}
|
||||
self.failures = 0
|
||||
self.retry_at = 0.0
|
||||
@@ -103,18 +169,22 @@ class Watch:
|
||||
|
||||
|
||||
class Link:
|
||||
__slots__ = ('entity', 'box', 'still', 'etag', 'used', 'lifetime')
|
||||
__slots__ = ('entity', 'box', 'still', 'etag', 'used', 'lifetime', 'cover')
|
||||
|
||||
def __init__(self, entity, box, now, still=None):
|
||||
def __init__(self, entity, box, now, still=None, cover=None):
|
||||
self.entity, self.box, self.used = entity, box, now
|
||||
self.still = still
|
||||
self.cover = cover # (size, background) of a media player's cover, else None
|
||||
self.etag = f'"{hashlib.sha1(still).hexdigest()[:16]}"' if still else ''
|
||||
self.lifetime = STILL_SECONDS if still else LINK_SECONDS
|
||||
|
||||
|
||||
class CameraFeed:
|
||||
def __init__(self, fetch, clock=time.monotonic):
|
||||
def __init__(self, fetch, clock=time.monotonic, fetch_cover=None, picture=None):
|
||||
"""`fetch(entity)` gives a camera's snapshot; `fetch_cover(entity)` a media player's picture and
|
||||
`picture(entity)` the address that picture has in its state right now ('' without one)."""
|
||||
self.fetch, self.clock = fetch, clock
|
||||
self.fetch_cover, self.picture = fetch_cover, picture
|
||||
self.watches, self.links = {}, {}
|
||||
|
||||
# ----- fetching -----
|
||||
@@ -194,6 +264,61 @@ class CameraFeed:
|
||||
cached = watch.frames[box] = (digest, image)
|
||||
return f'"{digest[:16]}-{box[0]}x{box[1]}"', cached[1]
|
||||
|
||||
# ----- covers -----
|
||||
# A media player's picture is fetched when a screen loads its cover link and Home Assistant's picture is another
|
||||
# one than the last fetched (or none was fetched yet): a cover never changes on its own between two loads, and the
|
||||
# screen loads again only when the state's picture mark changed. So a load waits for the picture of this moment.
|
||||
async def fetch_cover_one(self, entity, watch, picture):
|
||||
try:
|
||||
raw = await asyncio.wait_for(self.fetch_cover(entity), FETCH_SECONDS)
|
||||
if not raw:
|
||||
raise ValueError('empty picture')
|
||||
digest = hashlib.sha1(raw).hexdigest()
|
||||
if digest != watch.digest:
|
||||
watch.raw, watch.digest, watch.frames = raw, digest, {}
|
||||
watch.picture = picture
|
||||
if watch.failures:
|
||||
LOG.info('The cover of %s comes again', entity)
|
||||
watch.failures = 0
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as error:
|
||||
watch.failures += 1
|
||||
if watch.failures in (1, 30):
|
||||
LOG.info('No cover from %s (%s)', entity, type(error).__name__)
|
||||
|
||||
async def cover(self, entity, size, background, wait=FIRST_FRAME_SECONDS):
|
||||
"""(etag, BMP) of the player's cover at `size` with `background` behind its corners, or None when the player
|
||||
shows no picture or it cannot be fetched."""
|
||||
if self.fetch_cover is None or self.picture is None:
|
||||
return None
|
||||
picture = self.picture(entity)
|
||||
if not picture:
|
||||
return None
|
||||
watch = self.watch(entity)
|
||||
if watch.picture != picture or watch.raw is None:
|
||||
if watch.task is None or watch.task.done() or watch.fetching != picture:
|
||||
if watch.task is not None and not watch.task.done():
|
||||
watch.task.cancel()
|
||||
watch.fetching = picture
|
||||
watch.task = asyncio.ensure_future(self.fetch_cover_one(entity, watch, picture))
|
||||
try:
|
||||
await asyncio.wait_for(asyncio.shield(watch.task), wait)
|
||||
except asyncio.TimeoutError:
|
||||
return None
|
||||
if watch.picture != picture or watch.raw is None:
|
||||
return None
|
||||
key = ('cover', size, background)
|
||||
cached = watch.frames.get(key)
|
||||
if cached is None or cached[0] != watch.digest:
|
||||
try:
|
||||
image = await asyncio.get_running_loop().run_in_executor(None, encode_cover, watch.raw, size, background)
|
||||
except Exception as error:
|
||||
LOG.info('The cover of %s cannot be read (%s)', entity, type(error).__name__)
|
||||
return None
|
||||
cached = watch.frames[key] = (watch.digest, image)
|
||||
return f'"{watch.digest[:16]}-c{size}-{background:06X}"', cached[1]
|
||||
|
||||
# ----- links -----
|
||||
def prune(self):
|
||||
now = self.clock()
|
||||
@@ -202,11 +327,12 @@ class CameraFeed:
|
||||
while len(self.links) >= MAX_LINKS:
|
||||
del self.links[min(self.links, key=lambda token: self.links[token].used)]
|
||||
|
||||
def link(self, entity, box, still=None):
|
||||
"""A new random token for one camera at one size; `still` makes it one fixed image (an alert's)."""
|
||||
def link(self, entity, box, still=None, cover=None):
|
||||
"""A new random token for one camera at one size; `still` makes it one fixed image (an alert's), `cover`
|
||||
(size, background) a media player's cover."""
|
||||
self.prune()
|
||||
token = secrets.token_urlsafe(18)
|
||||
self.links[token] = Link(entity, box, self.clock(), still)
|
||||
self.links[token] = Link(entity, box, self.clock(), still, cover)
|
||||
return token
|
||||
|
||||
async def serve(self, token, etag=None):
|
||||
@@ -219,7 +345,7 @@ class CameraFeed:
|
||||
link.used = now
|
||||
if link.still:
|
||||
return (304, None, link.etag) if etag == link.etag else (200, link.still, link.etag)
|
||||
found = await self.frame(link.entity, link.box)
|
||||
found = await self.cover(link.entity, *link.cover) if link.cover else await self.frame(link.entity, link.box)
|
||||
if found is None:
|
||||
return 503, None, ''
|
||||
tag, image = found
|
||||
|
||||
Reference in New Issue
Block a user