aiocoap-widgets

#!/usr/bin/env -S pipx run
# SPDX-FileCopyrightText: Christian Amsüss and the aiocoap contributors
#
# SPDX-License-Identifier: MIT

# /// script
# requires-python = ">= 3.10"
# dependencies = [
#   "aiocoap >= 0.4.8, < 0.5",
#   "cbor2",
#   "pygobject >= 3.49"
# ]
# ///

"""Graphical utility to represent typical CoAP interfaces"""

import argparse

from aiocoap import resource
from aiocoap import Context
from aiocoap.resourcedirectory.client.register import Registerer

import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
import asyncio
from gi.events import GLibEventLoopPolicy

from widgets_common.gobject_resource import *


class Slider(FloatResource, GobjectBacked):
    if_ = 'core.s'
    unit = 'degC' # FIXME that should be somewhere in core

    widget_property = 'value'

    def __init__(self, args):
        self.widget = Gtk.Scale()
        self.widget.props.draw_value = False

        self.backend_widget = self.widget.props.adjustment = Gtk.Adjustment()

        self.backend_widget.props.lower = -10
        self.backend_widget.props.upper = 40
        self.backend_widget.props.value = 20

        super().__init__()

class WindowTitle(StringResource, GobjectBacked):
    if_ = 'core.p'

    widget_property = 'title'

    def __init__(self, window):
        self.backend_widget = self.widget = window

        super().__init__()

class ArgumentParser(argparse.ArgumentParser):
    def __init__(self):
        super().__init__(description=__doc__.split("\n")[0])

        self.add_argument('--anyport', help="Bind to a non-default port (useful for multiple instances on one computer, and in connection with --register-at)", action='store_true')
        self.add_argument('--register', help="Resource directory to register at (uses multicast if not specified)", metavar='RD', nargs='?', action='append')
        self.add_argument('--register-as', help="Endpoint name (or name.domain) to register as at the resource directory (Default: %(default)s)", default='widget', metavar='EP')

async def start_context(site, args):
    if args.anyport:
        ctx = await Context.create_client_context()
        ctx.serversite = site
    else:
        ctx = await Context.create_server_context(site)

    if args.register:
        ep, _, d = args.register_as.partition('.')
        registerer = Registerer(ctx, args.register[0], lt=120, registration_parameters=dict(ep=ep, d=d))

        async def shutdown():
            await registerer.shutdown()
            await ctx.shutdown()

        return shutdown
    else:
        return ctx.shutdown

class WidgetApplication(Gtk.Application):
    def __init__(self):
        super().__init__()
        self.hold()
        self.connect("activate", lambda s: asyncio.create_task(self.start()))

    async def start(self):
        p = ArgumentParser()
        args = p.parse_args()

        win = Gtk.Window()
        win.connect("delete-event", lambda *args: self.release())

        site = resource.Site()
        site.add_resource(['.well-known', 'core'], resource.WKCResource(site.get_resources_as_linkheader))
        site.add_resource(['window-title'], WindowTitle(win))

        # temporary show-all
        box = Gtk.VBox()
        win.add(box)

        rgb = RGBLight(None)
        box.add(rgb.widget)
        site.add_resource(['spot'], rgb)

        switch = Switch(None)
        box.add(switch.widget)
        site.add_resource(['switch'], switch)

        bulb = Bulb(None)
        box.add(bulb.widget)
        site.add_resource(['bulb'], bulb)

        slider = Slider(None)
        box.add(slider.widget)
        site.add_resource(['slider'], slider)

        bigbatch = SubsiteBatch(site)
        site.add_resource([], bigbatch)

        win.show_all()

        self.shutdown = await start_context(site, args)

def sync_main():
    asyncio.set_event_loop_policy(GLibEventLoopPolicy())

    app = WidgetApplication()
    app.run()

if __name__ == "__main__":
    sync_main()