Update state from external action

[tmarrinan]

Hello!
I am building a Trame app that will get data not only from the user interacting with the web page, but also from external sources (e.g. a second process or a remote service).
I have tried updating the state object from a different thread or from an asynchronous function, but the state does not seem to update (at least the web page does not reflect the change).

The following are examples of what I've tried:

server = get_server(client_type="vue2")
state = server.state
asynchronous.create_task(updateStateTest(state))

# ...

async def updateStateTest(state):
    await asyncio.sleep(10)
    print('changing state now...')
    state.test_val = 'Option B'

And:

thread = Thread(target=updateStateTest, args=(state,))
thread.start()

# ...

def updateStateTest(state):
    time.sleep(10)
    print('changing state now...')
    state.test_val = 'Option B'

I just have test_val set up in the Vue model as a select dropdown menu:

with SinglePageLayout(server) as layout:
        layout.title.set_text('Trame-Test')
        with layout.toolbar:
            vuetify.VSpacer()
            vuetify.VSelect(
                label='Test Menu',
                v_model=('test_val', 'Option A'),
                items=('[\'Option A\', \'Option B\', \'Option C\']',),
                hide_details=True,
                dense=True
            )

But "Option A" remains selected even after the "changing state now..." action takes place.

[tmarrinan]

Nevermind - I found the solution!

I need to add state.flush() after updating it

[tmarrinan]

OK - actually only halfway there. state.flush() works for asynchronous but not for a separate thread.
Since my external data will be coming from services that having blocking IO, running just via asynchronous blocks and never even ends up starting the Trame server.

[tmarrinan]

I now have something that works... it relies on a separate queue system rather than working with the state directly, so not sure if this is a real solution or more of a workaround, but it achieves what I need. Posting in case others run into similar situation:

import asyncio
from multiprocessing import Process, Queue
from multiprocessing.managers import BaseManager
from trame.app import get_server, asynchronous
from trame.widgets import vuetify
from trame.ui.vuetify import SinglePageLayout

class QueueManager(BaseManager):
    pass

def main():
    # create queue for Trame state
    state_queue = Queue()
    
    # start Trame app as new process
    trame_proc = Process(target=runTrameServer, args=(state_queue,))
    trame_proc.daemon = True
    trame_proc.start()

    # create queue for external data
    queue_data = Queue()

    # start Queue Manager as new process
    queue_mgr_proc = Process(target=runQueueManager, args=(queue_data))
    queue_mgr_proc.daemon = True
    queue_mgr_proc.start()

    # read data from external sources
    data = ''
    while data != 'quit':
        data = queue_data.get()
        state_queue.put({'test_val': data})

def runTrameServer(state_queue):
    # set up Trame application
    server = get_server(client_type="vue2")
    state = server.state
    ctrl = server.controller
    
    # callback for test value change
    def uiStateTestValUpdate(test_val, **kwargs):
        print(f'Test value changed: {test_val}.')

    # register callback
    state.change('test_val')(uiStateTestValUpdate)
    
    asynchronous.create_task(checkForStateUpdates(state, state_queue))
    
    # define webpage layout
    with SinglePageLayout(server) as layout:
        layout.title.set_text('Ascent-Trame')
        with layout.toolbar:
            vuetify.VSpacer()
            vuetify.VSelect(
                label='Test Menu 1',
                v_model=('test_val', 'Option A'),
                items=('[\'Option A\', \'Option B\', \'Option C\']',),
                hide_details=True,
                dense=True
            )

    # start Trame server
    server.start()

async def checkForStateUpdates(state, state_queue):
    while True:
        try:
            updated_state_values = state_queue.get(block=False)
            state.update(updated_state_values)
            state.flush()
        except:
            pass
        await asyncio.sleep(0)

def runQueueManager(queue_data):
    # register queues with Queue Manager
    QueueManager.register('get_data_queue', callable=lambda:queue_data)
    
    # create Queue Manager
    mgr = QueueManager(address=('127.0.0.1', 8000), authkey=b'trame')
    
    # start Queue Manager server
    server = mgr.get_server()
    server.serve_forever()

if __name__ == '__main__':
    main()

Data can then be pushed to this application by connecting to the QueueManager and putting data (in this instance, data should be 'Option A', 'Option B', or 'Option C' to properly update the test_val.

[jourdain]

That example illustrates a thread synchronizing method call to the event loop.

Here is another one doing state update from another thread.