Add asyncio call from gui #1576

atitoff · 2025-01-12T03:18:49Z

I wrote a simple library.
Asynchrony is launched in a separate thread.

import webview
import asyncio

from lib.wv_async import WVAsync, JsApi

# add user function
class Js(JsApi):
    def call23(self, d):
        print('call23', d)

html = """<!DOCTYPE html><html><head lang="en"><meta charset="UTF-8"></head><body>

<button onClick="pywebview.api.call('random', {a: 'ss', b: 25})">Test</button><br><br>
<button onClick="pywebview.api.call23({a: 'ss', b: 25})">Test2</button>

</body></html>"""

async def random(d):
    print('def random:', d)
    i = 0
    while True:
        await asyncio.sleep(1)
        print('random', i)
        i += 1


wv_app = WVAsync()
js_api = Js(wv_app.jq)

wv_app.registry('random', random)

window = webview.create_window('JS API example', html=html, js_api=js_api)
wv_app.start(window)
webview.start()

lib wv_async.py

import asyncio
from threading import Thread

import janus


class WVAsync:
    def __init__(self):
        self.jq = janus.Queue()
        self.js_api = JsApi(self.jq)
        self.window = None
        self._t = Thread(target=self._main)
        self._reg = {}  # {fn_name: fn}

    def registry(self, name, fn):
        self._reg[name] = fn

    def _on_closing(self):
        print('closing')
        self.jq.sync_q.put_nowait({'closing': True})

    def start(self, window):
        self.window = window
        self.window.events.closing += self._on_closing
        self._t.start()

    def _main(self):
        asyncio.run(self._main_loop())

    async def _main_loop(self):
        while True:  # ret = {name: {value}}
            ret = await self.jq.async_q.get()
            print(ret)
            for fn_name in ret:
                if fn_name in self._reg:
                    asyncio.create_task(self._reg[fn_name](ret[fn_name]))
                    await asyncio.sleep(0)
            if 'closing' in ret:
                break


class JsApi:
    def __init__(self, jq):
        self.jq = jq

    def call(self, rpc_name, d=None):
        print(rpc_name, d)
        self.jq.sync_q.put_nowait({rpc_name: d})

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add asyncio call from gui #1576

Add asyncio call from gui #1576

atitoff commented Jan 12, 2025 •

edited

Loading

Add asyncio call from gui #1576

Add asyncio call from gui #1576

Comments

atitoff commented Jan 12, 2025 • edited Loading

atitoff commented Jan 12, 2025 •

edited

Loading