Source code for caspia.pan.reactive.display

import asyncio
from binascii import b2a_base64
from zlib import compress

import caspia.reactive as r
from caspia.pan.reactive import lambda_


def display_update(display, render, inputs=()):
    async def display_buffer_changed(display_buffer):
        # compress and encode the buffer
        data = b2a_base64(compress(display_buffer)).decode('ascii')
        # update the display
        await display.update.write({
            'fill': 0,
            'show': True,
            'data': data,
            'format': 'HLSB',
        })

    async def at_startup(display_buffer_o):
        await asyncio.sleep(1.0)
        try:
            display_buffer = await display_buffer_o.observe()
            await display_buffer_changed(display_buffer)
        except r.errors.ValueNotReady:
            pass

    display_buffer = lambda_(render, *inputs)
    rule = display_buffer.do(display_buffer_changed)  # pylint: disable=no-member
    rule.name = f'Render display {display}'
    asyncio.ensure_future(at_startup(display_buffer))