About custom events

[Yurii-huang]

How do I customize an event to update the UI?
Also how do I deliver an event?

[Yurii-huang]

see https://textual.textualize.io/guide/events/, chapter "custom messages", do deliver event(call API)

[davep]

Does this mean you've answered your own question, or is there something you still need help with? (that document covers both custom messages and sending messages)

[Yurii-huang]

I want in thread post a message, the app instance is main thread, callback method is impossible

How to do?

class DiskWatcher(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        context = pyudev.Context()
        monitor = pyudev.Monitor.from_netlink(context)
        monitor.filter_by('block')
        for device in iter(monitor.poll, None):
            action = device.action
            device_node = device.device_node
            app.post_message(DiskTree.DiskUpdate(device.action, device.device_node))

class DiskTree(Tree):

    class DiskUpdate(Message):
        """ action: disk action with str|remove|add """
        def __init__(self, action, device_node):
            self.action = action
            self.device_node = device_node
            super().__init__()
        
    def __init__(self):
        super().__init__("block")
        self.root.expand()
        context = pyudev.Context()
        app.run_async
        dParentNode = None
        for device in context.list_devices(subsystem="block"):
            dParent = device.parent
            if dParent.device_type == None:
                dType = device.device_type
                dNode = device.device_node
                dParentNode = self.root.add(device.sys_name + " " + dType + " " + dNode, expand=True)
            else:
                if dParentNode:
                    dType = device.device_type
                    dNode = device.device_node
                    dPartType = device.get("ID_FS_TYPE")
                    if dPartType:
                        dParentNode.add_leaf(device.sys_name + " " + dType + " " + dNode + " " + dPartType)
        self.diskWatcher = DiskWatcher()
        self.diskWatcher.start()

    def on_disk_update(self, message: DiskUpdate) -> None:
        pass

class TabbedDemo(App):
    CSS = """
    Screen {
        background: darkblue;
        border: solid
    }

    RadioSet {
        align: left middle;
        width: 45%;
        border: solid
    }

    Label {
        border: solid;
        color: $text-disabled;
    }
    """

    def compose(self) -> ComposeResult:
        with TableView():
            with RadioSet():
                yield RadioButtonCos("English", True)
                yield RadioButtonCos("Chinese")
            yield Markdown(C)
            yield DiskTree()

    def on_mount(self) -> None:
        pass

if __name__ == "__main__":
    app = TabbedDemo()
    app.run()

[davep]

Have you read up on Textual workers?

[Yurii-huang]

I really didn't see that the app can directly start asynchronous threads. But it seems that this is not what I want. I want the features of each component to be separated, rather than written into the app instance. This is more beneficial for code structure. This has now been resolved, and it can be achieved through both built-in workers and native threads.

Now there is a new question. If I want to send events to a specific class, I can only obtain objects by querying the component ID through the app, right?

[Yurii-huang]

first register a handler method. If U Message definitions with:

class DiskTree:
    class DiskUpdate:
    ...

then method name use "on_disk_tree_disk_update"

before send message, query widget class, call class instance post_ message method can accurately deliver information to a certain class

class DiskWatcher(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        context = pyudev.Context()
        monitor = pyudev.Monitor.from_netlink(context)
        monitor.filter_by('block')
        for device in iter(monitor.poll, None):
            action = device.action
            device_node = device.device_node
            diskTree = app.query_one("#DiskTree")
            diskTree.post_message(
                DiskTree.DiskUpdate(device.action, device.device_node))

...

class DiskTree(Tree):

    class DiskUpdate(Message):
        """ action: disk action with str|remove|add """
        def __init__(self, action, device_node):
            self.action = action
            self.device_node = device_node
            super().__init__()
        def __init__(self):
            super().__init__("block", id="DiskTree")
            self.diskWatcher = DiskWatcher()
            self.diskWatcher.start()

...

    def on_disk_tree_disk_update(self, message: DiskUpdate) -> None:
        pass