Send pubsub broadcast message from outside flet

[midiwidi]

Question

I'm building a flet application which will be used to command a hardware device and display data received from the device.

In addition to a desktop application I would also like to give multiple users access using a web app with multiple sessions.
I have a class instance in a global scope which interfaces to the hardware. For commanding (sending data to the hardware) the sessions call the write function of the class using a lock for thread safety (blue arrows). For the opposite direction (receiving data from the hardware) I currently use a similar mechanism where a click event in the sessions calls a read function (again using a lock) which then requests data from the hardware and returns it to the sessions with the function return value.

Now I want to implement periodic data requests from the hardware. I don't want each session to periodically read data using the mechanism described above because that would send 3 data requests to the hardware, receive 3 replies and return them to the corresponding sessions (for 3 sessions as in the diagram). Instead I would like to have a periodic timer in my interface class (a periodic thread) which would send 1 request to the hardware, receive the reply and then broadcast the data to all sessions using pubsub (red arrows).

Is there a way I can use pubsub outside of the flet application (in the timer thread) to send a broadcast message to the sessions? Or could there be a headless session, a page which is not showing anything and is not starting a web server in which I can embed my interface class and use its page.pubsub.send_all() function to do the broadcast?
If possible I would like the same code to work as a desktop app.

Edit

Here is a minimalist example to help understand what I'm looking for

Code sample

import flet as ft
from datetime import datetime
import threading

class PeriodicThread(object):
    def __init__(self, callback=None, period=1, name=None, *args, **kwargs):
        self.name = name
        self.args = args
        self.kwargs = kwargs
        self.callback = callback
        self.period = period
        self.stop = False
        self.current_timer = None
        self.schedule_lock = threading.Lock()

    def set_period(self, period):
        self.period = period
        
    def start(self):
        self.stop = False
        self.schedule_timer()

    def run(self):
        if self.callback is not None:
            self.callback(*self.args, **self.kwargs)

    def _run(self):
        try:
            self.run()
        except Exception as e:
            pass
        finally:
            with self.schedule_lock:
                if not self.stop:
                    self.schedule_timer()

    def schedule_timer(self):
        self.current_timer = threading.Timer(self.period, self._run, *self.args, **self.kwargs)
        if self.name:
            self.current_timer.name = self.name
        self.current_timer.start()

    def cancel(self):
        with self.schedule_lock:
            self.stop = True
            if self.current_timer is not None:
                self.current_timer.cancel()

    def join(self):
        self.current_timer.join()


class hardware_interface():
    def __init__(self):
        self.lock = threading.Lock()
        self.cnt = 0
        self.timer = PeriodicThread(callback=self.cb_timer, period=1, name="timer_periodic_update")
        self.timer.start()
        
    def send_command(self, cmd, cmd_data):
        print(f'Instrument: executing command "{cmd} {cmd_data}"')
        
    def read_oneoff_data(self):
        data = self.cnt
        self.cnt += 1
        print(f'Instrument: data requested, replied {data}')
        return data
    
    def cb_timer(self):
        data = datetime.now()
        print(f'Instrument: data requested, replied {data}')
        #TODO: Send broadcast message with the data to all app sessions
        #      If this would be a flet page object I could use
        #      self.pubsub.send_all(data)


class App(ft.UserControl):
    def build(self):
        
        self.tf_cmd = ft.TextField(label="CMD")
        self.tf_cmd_data = ft.TextField(label="CMD Data")
        self.tf_req_data = ft.TextField(label="Requested Data")
        self.tf_per_data = ft.TextField(label="Periodic Data")
        
        page = ft.Column(
            [
                ft.Row(
                    [
                        self.tf_cmd,
                        self.tf_cmd_data,
                        ft.ElevatedButton(text="Send Command", on_click=self.btn_send_click)
                    ]
                ),        
                ft.Row(
                    [
                        self.tf_req_data,
                        ft.ElevatedButton(text="Request Data", on_click=self.btn_reqdata_click)
                    ]
                ),
                ft.Row(
                    [
                        self.tf_per_data
                    ]
                )
            ]
        )
        self.page.pubsub.subscribe(self.on_HW_periodic_data_msg)
        return page
    
    
    def btn_send_click(self, e):
        g_hw_if.lock.acquire(timeout=3)
        g_hw_if.send_command(self.tf_cmd.value, self.tf_cmd_data.value)
        g_hw_if.lock.release()
        
    def btn_reqdata_click(self, e):
        g_hw_if.lock.acquire(timeout=3)
        self.tf_req_data.value = g_hw_if.read_oneoff_data()
        g_hw_if.lock.release()
        self.update()
        
    def on_HW_periodic_data_msg(self, msg):
        self.tf_per_data.value = msg
        self.update()


def session_main(page: ft.Page):
    app = App()
    page.add(app)


def main():
    global g_hw_if

    g_hw_if = hardware_interface()
    ft.app(target=session_main, view=ft.WEB_BROWSER, port=45678)

if __name__ == '__main__':
    main()

If I run this code a tab with the web app will open in my browser. I open another tab and enter the same URL as in the first one to start a new session of the web app. Now I can send a command to the simulated instrument (class hardware_interface) by entering e.g. "loglevel" in the CMD textbox, "3" in the CMD Data textbox and then clicking the "Send command" button in either of the tabs. Similarly each session can request data from the simulated instrument by clicking the "Request Data" button. The class hardware_interface has a timer callback function which periodically requests data from the instrument. What I'm looking for is a way to send the data I get in the callback to all sessions of the flet app (at the #TODO comment). If the hardware_interface class would be a flet page then I could use self.pubsub.send_all(data) but it isn't.

Edit 2: My Solution

Triggered by the idea from @david-shu I came up with the following solution. For each flet session there is a receiver task running in a separate thread. It integrates a queue which is registered centrally in a sessions object. The timer thread which requests data from the hardware puts the data into the queues of all active sessions using the sessions object.

import flet as ft
from datetime import datetime
import threading
import queue

class data_receiver(threading.Thread):
    def __init__(self, callback, session_nr):
        global g_sessions
        
        super().__init__()
        self.cb = callback
        self.stop = False
        self.queue = queue.Queue()
        g_sessions.lock.acquire(timeout=1)
        g_sessions.receiver_queues[session_nr] = self.queue
        g_sessions.lock.release()
        
        
    def run(self):
        while not self.stop:
            if self.queue.not_empty:
                self.cb(self.queue.get())


class PeriodicThread(object):
    def __init__(self, callback=None, period=1, name=None, *args, **kwargs):
        self.name = name
        self.args = args
        self.kwargs = kwargs
        self.callback = callback
        self.period = period
        self.stop = False
        self.current_timer = None
        self.schedule_lock = threading.Lock()

    def set_period(self, period):
        self.period = period
        
    def start(self):
        self.stop = False
        self.schedule_timer()

    def run(self):
        if self.callback is not None:
            self.callback(*self.args, **self.kwargs)

    def _run(self):
        try:
            self.run()
        except Exception as e:
            print(e)
            pass
        finally:
            with self.schedule_lock:
                if not self.stop:
                    self.schedule_timer()

    def schedule_timer(self):
        self.current_timer = threading.Timer(self.period, self._run, *self.args, **self.kwargs)
        if self.name:
            self.current_timer.name = self.name
        self.current_timer.start()

    def cancel(self):
        with self.schedule_lock:
            self.stop = True
            if self.current_timer is not None:
                self.current_timer.cancel()

    def join(self):
        self.current_timer.join()


class hardware_interface():
    def __init__(self):
        self.lock = threading.Lock()
        self.cnt = 0
        self.timer = PeriodicThread(callback=self.cb_timer, period=1, name="timer_periodic_update")
        self.connect()
        
    def send_command(self, cmd, cmd_data):
        print(f'Instrument: executing command "{cmd} {cmd_data}"')
        
    def read_oneoff_data(self):
        data = self.cnt
        self.cnt += 1
        print(f'Instrument: data requested, replied {data}')
        return data
    
    def connect(self):
        self.timer.start()
        print('Instrument connected')
        
    def disconnect(self):
        self.timer.cancel()
        print('Instrument disconnected')
    
    def cb_timer(self):
        # Put the data received from the instrument into the receive queues of the sessions
        # In this example without a real instrument it is simulated by just getting the current time
        data = datetime.now()
        for receiver_queue in g_sessions.receiver_queues.values():
            receiver_queue.put(data)
                       

class App(ft.UserControl):
    def build(self):
        g_sessions.lock.acquire(timeout=1)
        self.session_nr = g_sessions.nr
        g_sessions.lock.release()
        
        self.tf_cmd = ft.TextField(label="CMD")
        self.tf_cmd_data = ft.TextField(label="CMD Data")
        self.tf_req_data = ft.TextField(label="Requested Data")
        self.tf_per_data = ft.TextField(label="Periodic Data")
        
        page = ft.Column(
            [
                ft.Row(
                    [
                        self.tf_cmd,
                        self.tf_cmd_data,
                        ft.ElevatedButton(text="Send Command", on_click=self.btn_send_click)
                    ]
                ),        
                ft.Row(
                    [
                        self.tf_req_data,
                        ft.ElevatedButton(text="Request Data", on_click=self.btn_reqdata_click)
                    ]
                ),
                ft.Row(
                    [
                        self.tf_per_data
                    ]
                )
            ]
        )
        
        self.dat_rcv = data_receiver(self.cb_data_receiver, self.session_nr)
        self.dat_rcv.start()
        return page
    
    
    def btn_send_click(self, e):
        g_hw_if.lock.acquire(timeout=3)
        g_hw_if.send_command(self.tf_cmd.value, self.tf_cmd_data.value)
        g_hw_if.lock.release()
        
    def btn_reqdata_click(self, e):
        g_hw_if.lock.acquire(timeout=3)
        self.tf_req_data.value = g_hw_if.read_oneoff_data()
        g_hw_if.lock.release()
        self.update()
        
    def cb_data_receiver(self, data):
        self.tf_per_data.value = data
        self.update()


class session():
    def __init__(self):
        self.count = 0
        self.nr = 0
        self.lock = threading.Lock()
        self.receiver_queues = {}


def session_disconnect(e):
    global g_hw_if 
    global g_sessions
    
    g_sessions.lock.acquire(timeout=1)
    if g_sessions.count > 0:
        session_nr = e.page.controls[0].session_nr
        e.page.controls[0].dat_rcv.stop = True
        print(f"Session {session_nr} closed")
        if session_nr in g_sessions.receiver_queues:
            del g_sessions.receiver_queues[session_nr]
        g_sessions.count = g_sessions.count - 1
        if g_sessions.count == 0:
            g_hw_if.lock.acquire(timeout=3)
            g_hw_if.disconnect()
            g_hw_if.lock.release()
    g_sessions.lock.release() 
    

def window_event(e):
    global g_hw_if
    
    if e.data == 'close':
        g_hw_if.lock.acquire(timeout=3)
        g_hw_if.disconnect()
        g_hw_if.lock.release()
        e.page.controls[0].dat_rcv.stop = True
        #e.page.controls[0].dat_rcv.join()
        e.page.window_destroy()
        
            
def session_main(page: ft.Page):
    global g_sessions
    
    g_sessions.lock.acquire(timeout=1)
    g_sessions.count = g_sessions.count + 1
    g_sessions.nr = g_sessions.nr + 1
    g_sessions.lock.release() 
    page.on_disconnect = session_disconnect
    page.window_prevent_close = True
    page.on_window_event = window_event
    
    app = App(page)
    page.add(app)
    print(f"Session {app.session_nr} started")


def main():
    global g_hw_if
    global g_sessions

    g_hw_if = hardware_interface()
    g_sessions = session()
    ft.app(target=session_main, view=ft.WEB_BROWSER, port=45678)
    #ft.app(target=session_main)

if __name__ == '__main__':
    main()

[midiwidi]

I've asked the question also on stackoverflow including example code. I will report here if I get help there and find a solution.

[FeodorFitsner]

I see what you mean. I recently needed a pubsub that sits on top of all sessions in my own app.
Let me check if it's something we could easily change.

[midiwidi]

@FeodorFitsner is the source code of this project on GitHub so that I could have a look? Or do you mind telling me which method you have used?

[midiwidi]

I would really appreciate if somebody had an idea I could investigate or could tell me that what I'm trying to do is not possible with flet.
I edited my original post and added some example code to make it clearer what I'm trying to achieve.

[david-shu]

You can try celery, put the messages generated by the hardware into the queue, and then listen for the queue message notification in a separate thread. If there is a notification of the successful completion of the task, call pubsub.send_all to broadcast the message to all flet clients

class CeleryAppMonitor( threading.Thread ):
    def __init__( self, capp, pubsub=None ):
        super().__Init__()
        self.capp = capp
        self.pubsub = pubsub
    def run( self ):
           def on_task_succeeded( event ):
                state = capp.events.State()
                state.event( event )
                eid = event.get( 'uuid' )
                task = state.tasks.get( eid )
                if task:
                        info = task.info()
                        if self.pubsub:
                             self.pubsub.send_all( json.dumps( info ) )

           capp=self.capp
           while 1:
                 with capp.connection() as conn:
                        recv = capp.events.Receiver( conn, handlers={ 'task-succeeded':on_task_succeed})
                        recv.capture( )

class FletApp( object ):
        monitor = None
        def __new__( cls, page ):
               if not cls.monitor:
                   from celery_app import app as capp   # your celery app
                   cls.monitor = CeleryAppMonitor( capp, pubsub=page.pubsub )
                   cls.monitor.start()
               return super().__new__(cls)

async def main( page ):
      app = FletApp( page )
    
def serve():
     app = FastAPI()
     fletapp = flet_fastapi.app( main )
     app.mount('/', fletapp )
     uvicorn.run( app )

[midiwidi]

Thanks @david-shu for your suggestion. I've never used Celery before and are trying to wrap my head around and understand it.
I tried to combine your code with my code and ended up with that

HWcomms.py

import flet as ft
from fastapi import FastAPI
import flet.fastapi as flet_fastapi
from datetime import datetime
import threading
import json
import uvicorn
from celery import Celery

capp = Celery('celery_app', broker='pyamqp://guest@localhost//')

@capp.task
def mycapp():
    print('executing celery_app')
    
    
class PeriodicThread(object):
    def __init__(self, callback=None, period=1, name=None, *args, **kwargs):
        self.name = name
        self.args = args
        self.kwargs = kwargs
        self.callback = callback
        self.period = period
        self.stop = False
        self.current_timer = None
        self.schedule_lock = threading.Lock()

    def set_period(self, period):
        self.period = period
        
    def start(self):
        self.stop = False
        self.schedule_timer()

    def run(self):
        if self.callback is not None:
            self.callback(*self.args, **self.kwargs)

    def _run(self):
        try:
            self.run()
        except Exception as e:
            print(e)
            pass
        finally:
            with self.schedule_lock:
                if not self.stop:
                    self.schedule_timer()

    def schedule_timer(self):
        self.current_timer = threading.Timer(self.period, self._run, *self.args, **self.kwargs)
        if self.name:
            self.current_timer.name = self.name
        self.current_timer.start()

    def cancel(self):
        with self.schedule_lock:
            self.stop = True
            if self.current_timer is not None:
                self.current_timer.cancel()

    def join(self):
        self.current_timer.join()


class hardware_interface():
    def __init__(self):
        self.lock = threading.Lock()
        self.cnt = 0
        self.timer = PeriodicThread(callback=self.cb_timer, period=1, name="timer_periodic_update")
        self.timer.start()
        
    def send_command(self, cmd, cmd_data):
        print(f'Instrument: executing command "{cmd} {cmd_data}"')
        
    def read_oneoff_data(self):
        data = self.cnt
        self.cnt += 1
        print(f'Instrument: data requested, replied {data}')
        return data
    
    def cb_timer(self):
        data = datetime.now()
        print(f'Instrument: data requested, replied {data}')

        #TODO: Send broadcast message with the data to all app sessions
        #      If this would be a flet page object I could use
        #      self.pubsub.send_all(data)
        mycapp.delay()



class CeleryAppMonitor( threading.Thread ):
    def __init__( self, capp, pubsub=None ):
        super().__init__()
        self.capp = capp
        self.pubsub = pubsub
    def run( self ):
        def on_task_succeeded( event ):
            state = capp.events.State()
            state.event( event )
            eid = event.get( 'uuid' )
            task = state.tasks.get( eid )
            if task:
                info = task.info()
                if self.pubsub:
                    self.pubsub.send_all( json.dumps( info ) )
        
        capp=self.capp
        while 1:
            with capp.connection() as conn:
                recv = capp.events.Receiver( conn, handlers={ 'task-succeeded':on_task_succeeded})
                recv.capture( )
                        

class App(ft.UserControl):
    monitor = None
    def __new__( cls, page ):
        if not cls.monitor:
            cls.monitor = CeleryAppMonitor( capp, pubsub=page.pubsub )
            cls.monitor.start()
        return super().__new__(cls)
    
    def build(self):
        
        self.tf_cmd = ft.TextField(label="CMD")
        self.tf_cmd_data = ft.TextField(label="CMD Data")
        self.tf_req_data = ft.TextField(label="Requested Data")
        self.tf_per_data = ft.TextField(label="Periodic Data")
        
        page = ft.Column(
            [
                ft.Row(
                    [
                        self.tf_cmd,
                        self.tf_cmd_data,
                        ft.ElevatedButton(text="Send Command", on_click=self.btn_send_click)
                    ]
                ),        
                ft.Row(
                    [
                        self.tf_req_data,
                        ft.ElevatedButton(text="Request Data", on_click=self.btn_reqdata_click)
                    ]
                ),
                ft.Row(
                    [
                        self.tf_per_data
                    ]
                )
            ]
        )
        self.page.pubsub.subscribe(self.on_HW_periodic_data_msg)
        return page
    
    
    def btn_send_click(self, e):
        g_hw_if.lock.acquire(timeout=3)
        g_hw_if.send_command(self.tf_cmd.value, self.tf_cmd_data.value)
        g_hw_if.lock.release()
        
    def btn_reqdata_click(self, e):
        g_hw_if.lock.acquire(timeout=3)
        self.tf_req_data.value = g_hw_if.read_oneoff_data()
        g_hw_if.lock.release()
        self.update()
        
    def on_HW_periodic_data_msg(self, msg):
        self.tf_per_data.value = msg
        self.update()


async def session_main(page: ft.Page):
    app = App(page)
    page.add(app)


def main():
    global g_hw_if

    g_hw_if = hardware_interface()
    #ft.app(target=session_main, view=ft.WEB_BROWSER, port=45678)
    app = FastAPI()
    fletapp = flet_fastapi.app( session_main )
    app.mount('/', fletapp )
    uvicorn.run( app )

if __name__ == '__main__':
    main()

It runs without errors. I installed RabbitMQ as the broker.

I have several questions

• How can I put the data generated by my hardware into the queue and have the on_task_succeeded function triggered? I'm calling mycapp.delay() in my timer callback but I don't see the text "executing celery_app" be printed neither is the on_task_succeeded() function called.
• I'm trying to understand the structure. Is it correct that there is a CeleryAppMonitor for each flet session (so 3 monitors in case of the 3 sessions in my example)? If I put data in the queue, the on_task_succeeded handlers of all sessions will be triggered and then each session sends a pubsub broadcast message to all other sessions? So a session would receive pubsub messages with the same content from different other sessions?
• If there is 1 monitor per session, why should I use self.pubsub.send_all then. Couldn't I directly update the data of the session?
• Isn't Celery + RabbitMQ basically doing the same thing as pubsub? If an external service like RabbitMQ is required anyway could I use something simpler than Celery such as MQTT and use that instead of pubsub? Basically subscribe each session to a MQTT topic and publish the data from my hardware to that topic?
• Is there a way of doing that all internally without the need for an external service / broker such as RabbitMQ or MQTT?

[midiwidi]

I found a solution for my problem which can be found in my original post at the top. It works as intended. There is only one problem left and maybe somebody here knows how to fix it.
If I run the application as a desktop app by changing

ft.app(target=session_main, view=ft.WEB_BROWSER, port=45678)
#ft.app(target=session_main)

to

#ft.app(target=session_main, view=ft.WEB_BROWSER, port=45678)
ft.app(target=session_main)

then I get the following error when I close the GUI window

Exception in callback BaseEventLoop.run_in_executor(<concurrent.f...001F6FF3BBCB0>, <function ses...001F6FD2B37E0>, <flet_core.co...001F6FD4313A0>)
handle: <Handle BaseEventLoop.run_in_executor(<concurrent.f...001F6FF3BBCB0>, <function ses...001F6FD2B37E0>, <flet_core.co...001F6FD4313A0>)>
Traceback (most recent call last):
  File "C:\Users\username\AppData\Local\Programs\Python\Python312\Lib\asyncio\events.py", line 84, in _run
    self._context.run(self._callback, *self._args)
  File "C:\Users\username\AppData\Local\Programs\Python\Python312\Lib\asyncio\base_events.py", line 860, in run_in_executor
    executor.submit(func, *args), loop=self)
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\username\AppData\Local\Programs\Python\Python312\Lib\concurrent\futures\thread.py", line 170, in submit
    raise RuntimeError('cannot schedule new futures after shutdown')
RuntimeError: cannot schedule new futures after shutdown

[burhansvural]

import flet as ft
from datetime import datetime
import threading
import queue
from dataclasses import dataclass


class DataReceiver(threading.Thread):
    def __init__(self, callback, session_nr):
        global g_sessions

        super().__init__()
        self.cb = callback
        self.stop = False
        self.queue = queue.Queue()
        g_sessions.lock.acquire(timeout=1)
        g_sessions.receiver_queues[session_nr] = self.queue
        g_sessions.lock.release()

    def run(self):
        while not self.stop:
            if self.queue.not_empty:
                self.cb(self.queue.get())

    def __del__(self):
        global g_sessions
        g_sessions = None


class PeriodicThread(object):
    def __init__(self, callback=None, period=1, name=None, *args, **kwargs):
        self.name = name
        self.args = args
        self.kwargs = kwargs
        self.callback = callback
        self.period = period
        self.stop = False
        self.current_timer = None
        self.schedule_lock = threading.Lock()

    def set_period(self, period):
        self.period = period

    def start(self):
        self.stop = False
        self.schedule_timer()

    def run(self):
        if self.callback is not None:
            self.callback(*self.args, **self.kwargs)

    def _run(self):
        try:
            self.run()
        except Exception as e:
            print(e)
            pass
        finally:
            try:
                with self.schedule_lock:
                    if not self.stop:
                        self.schedule_timer()
            except Exception as err:
                pass

    def schedule_timer(self):
        try:
            self.current_timer = threading.Timer(self.period, self._run, *self.args, **self.kwargs)
            if self.name:
                self.current_timer.name = self.name
            self.current_timer.start()
        except Exception as err:
            pass

    def cancel(self):
        with self.schedule_lock:
            self.stop = True
            if self.current_timer is not None:
                self.current_timer.cancel()

    def join(self):
        self.current_timer.join()

    def __del__(self):
        self.stop = False
        with self.schedule_lock:
            self.stop = True
            if self.current_timer is not None:
                self.current_timer.cancel()
                self.current_timer = None
        self.schedule_lock = None


class HardwareInterface:
    def __init__(self):
        self.lock = threading.Lock()
        self.cnt = 0
        self.timer = PeriodicThread(callback=self.cb_timer, period=1, name="timer_periodic_update")
        self.connect()

    def send_command(self, cmd, cmd_data):
        print(f'Instrument: executing command "{cmd} {cmd_data}"')

    def read_oneoff_data(self):
        data = self.cnt
        self.cnt += 1
        print(f'Instrument: data requested, replied {data}')
        return data

    def connect(self):
        self.timer.start()
        print('Instrument connected')

    def disconnect(self):
        self.timer.cancel()
        print('Instrument disconnected')

    def cb_timer(self):
        # Put the data received from the instrument into the receive queues of the sessions
        # In this example without a real instrument it is simulated by just getting the current time
        data = datetime.now()
        for receiver_queue in g_sessions.receiver_queues.values():
            receiver_queue.put(data)

    def __del__(self):
        self.lock.release()
        self.lock = None
        self.timer.cancel()
        self.timer = None


class App(ft.Column):
    def __init__(self, *args, **kwargs):
        super().__init__()
        self.args = args
        self.kwargs = kwargs
        self.tf_cmd = ft.TextField(label="CMD")
        self.tf_cmd_data = ft.TextField(label="CMD Data")
        self.tf_req_data = ft.TextField(label="Requested Data")
        self.tf_per_data = ft.TextField(label="Periodic Data")
        self.session_nr = g_sessions.nr
        self.dat_rcv = DataReceiver(self.cb_data_receiver, self.session_nr)

    def build(self):
        global g_sessions
        g_sessions.lock.acquire(timeout=1)
        g_sessions.lock.release()

        page = ft.Column(
            [
                ft.Row(
                    [
                        self.tf_cmd,
                        self.tf_cmd_data,
                        ft.ElevatedButton(text="Send Command", on_click=self.btn_send_click)
                    ]
                ),
                ft.Row(
                    [
                        self.tf_req_data,
                        ft.ElevatedButton(text="Request Data", on_click=self.btn_reqdata_click)
                    ]
                ),
                ft.Row(
                    [
                        self.tf_per_data
                    ]
                )
            ]
        )

        self.dat_rcv.start()
        return page

    def btn_send_click(self, e):
        global g_hw_if
        g_hw_if.lock.acquire(timeout=3)
        g_hw_if.send_command(self.tf_cmd.value, self.tf_cmd_data.value)
        g_hw_if.lock.release()

    def btn_reqdata_click(self, e):
        global g_hw_if
        g_hw_if.lock.acquire(timeout=3)
        self.tf_req_data.value = g_hw_if.read_oneoff_data()
        g_hw_if.lock.release()
        self.update()

    def cb_data_receiver(self, data):
        self.tf_per_data.value = data
        self.update()


@dataclass
class Session:
    def __init__(self):
        self.count = 0
        self.nr = 0
        self.lock = threading.Lock()
        self.receiver_queues = {}


def session_disconnect(e):
    global g_hw_if
    global g_sessions

    g_sessions.lock.acquire(timeout=1)
    if g_sessions.count > 0:
        session_nr = e.page.controls[0].session_nr
        e.page.controls[0].dat_rcv.stop = True
        print(f"Session {session_nr} closed")
        if session_nr in g_sessions.receiver_queues:
            del g_sessions.receiver_queues[session_nr]
        g_sessions.count = g_sessions.count - 1
        if g_sessions.count == 0:
            g_hw_if.lock.acquire(timeout=3)
            g_hw_if.disconnect()
            g_hw_if.lock.release()
    g_sessions.lock.release()


def window_event(e):
    global g_hw_if

    if e.data == 'close':
        g_hw_if.lock.acquire(timeout=3)
        g_hw_if.disconnect()
        g_hw_if.lock.release()
        e.page.controls[0].dat_rcv.stop = True
        # e.page.controls[0].dat_rcv.join()
        e.page.window_destroy()


def session_main(page: ft.Page):
    global g_sessions

    g_sessions.lock.acquire(timeout=1)
    g_sessions.count = g_sessions.count + 1
    g_sessions.nr = g_sessions.nr + 1
    g_sessions.lock.release()
    page.on_disconnect = session_disconnect
    page.window_prevent_close = True
    page.on_window_event = window_event

    app = App(page)
    page.add(app)
    print(f"Session {app.session_nr} started")


g_hw_if = None
g_sessions = None


def main():
    global g_hw_if
    global g_sessions

    g_hw_if = HardwareInterface()
    g_sessions = Session()
    


if __name__ == '__main__':
    main()
    # ft.app(target=session_main)
    ft.app(target=session_main, view=ft.WEB_BROWSER, port=45678)