Dynamically update Datatable

[martibc16]

Hello, I'm starting to use Textual to create a simple TUI for an existing aplication.

I basically need to show a Table that will have its contents changed based on external triggers from the main app.

For the moment I have the following:

from textual.app import App, ComposeResult
from textual.widgets import Button, Footer, Header, Static, DataTable

COLUMNS=("UUID", "Registros", "Paquetes descargados", "Registros decargados", "Progreso actual")

class GatewayApp(App):
    """A Textual app to manage stopwatches."""

    BINDINGS = [("d", "toggle_dark", "Toggle dark mode")]

    def compose(self) -> ComposeResult:
        """Create child widgets for the app."""
        yield Header()
        yield Footer()
        yield DataTable()

    def on_mount(self) -> None:
        self.table = self.query_one(DataTable)
        self.table.add_columns(*COLUMNS)
        
            
    def new_device(self, uuid : int, regs : int) -> None:
        cells=(uuid, regs, 0, 0, 'N/D')
        self.table.add_row(*cells, key= uuid)
        
    def action_toggle_dark(self) -> None:
        """An action to toggle dark mode."""
        self.dark = not self.dark


if __name__ == "__main__":
    app = GatewayApp()
    app.run()
    
    app.new_device(150, 600)
    app.new_device(500, 230)

This gets me:

The problem is that it seems like the calls to ne_device() are not taking effect. However, if I move this calls to on_mount like so:

    def on_mount(self) -> None:
        self.table = self.query_one(DataTable)
        self.table.add_columns(*COLUMNS)
        
        self.new_device(150, 600)
        self.new_device(500, 230)

It works:

So, I don't understand what's the difference between both cases.

i obviously need the first case. Because new_device will be called from outside when a change happens in the main app.

[davep]

The difference is that, in the first case, you're making the calls to new_device after your application has finished running and it has closed down.

[martibc16]

Hi @davep and thanks for the quick response.

It makes sense. Does this mean that run() is blocking and it will only return when the aplication finishes? (Namely when I hit ctrl+c).

If this is true then I would have to make the calls to new_device() from another thread when app.run() has already been called right?

[TomJGooding]

Textual is powered by the asyncio framework, the run method (I think) starts that event loop.

You'll probably find the Worker API docs useful here (possibly also how to pass arguments to the app).

[martibc16]

EDIT: See below answer before

Hi @TomJGooding,

The worker API makes calls from inside the Textual module. I need to call methods inside the GatewayApp from the outside so that the TUI canm be updated with the latest information comming from the main app.

The thing is that this main app has already been developed and it is working wonderfully. However, it was designed to work independently with no user feedback. Now they are asking me to show in a graphical way what the app is doing. So the idea is that the main app can call the Textual modules with the new rows it has to the Datatable.

Correct me if I'm wrong, but I don't think the Worker API could be of help in this situation.

So, the general question would be: How can I update information on a DataTable calling methods from the outside?

[martibc16]

A little update:

from textual.app import App, ComposeResult
from textual.widgets import Button, Footer, Header, Static, DataTable
COLUMNS=("UUID", "Registros", "Paquetes descargados", "Registros decargados", "Progreso actual", "Descarga activa")
import threading, time

class GatewayApp(App):
    """A Textual app to manage stopwatches."""

    BINDINGS = [("d", "toggle_dark", "Toggle dark mode"),
                ("f", "table_cleanup", "Quitar dispositivos inactivos")]

    def compose(self) -> ComposeResult:
        """Create child widgets for the app."""
        yield Header()
        yield Footer()
        yield DataTable()

    def on_mount(self) -> None:
        table = self.query_one(DataTable)
        table.add_columns(*COLUMNS)
        
        # self.new_device(150, 600)
        # self.new_device(500, 230)
                
    def new_device(self, uuid : int, regs : int) -> None:
        table = self.query_one(DataTable)
        cells=(uuid, regs, 0, 0, 'N/D', 'No')
        table.add_row(*cells, key= uuid)
        
    def action_toggle_dark(self) -> None:
        """An action to toggle dark mode."""
        self.dark = not self.dark
        
    def action_table_cleanup(self) -> None:
        a=0
        while True:
            
            try:
                if self.table.get_cell_at((a, 5)) == "No":
                    self.table.remove_row(self.table.coordinate_to_cell_key((a, 5)))
                a=a+1
            except Exception:
                break
            
def test(a):
    time.sleep(5)
    app.new_device(54, 600)
    app.new_device(500, 230)
    
if __name__ == "__main__":
    app = GatewayApp()
    threading.Thread(target=test, args=(1,)).start()
    app.run()

This works as expected. However, I don't like that I have to time.sleep the thread for some time to make sure the app.run has run completely.

I supose I would need some kind of semaphore to signal when the app has finished running the on_mount method....

[willmcgugan]

You should avoid using thread directly with Textual apps. It is far better to use threaded workers which will allow you to manage the lifetime of the thread (no need for sleep).

Here's your code modified to use a threaded worker:

from textual.app import App, ComposeResult
from textual import work
from textual.widgets import Button, Footer, Header, Static, DataTable

import threading, time

COLUMNS = (
    "UUID",
    "Registros",
    "Paquetes descargados",
    "Registros decargados",
    "Progreso actual",
    "Descarga activa",
)


class GatewayApp(App):
    """A Textual app to manage stopwatches."""

    BINDINGS = [
        ("d", "toggle_dark", "Toggle dark mode"),
    ]

    def compose(self) -> ComposeResult:
        """Create child widgets for the app."""
        yield Header()
        yield Footer()
        yield DataTable()

    def on_mount(self) -> None:
        table = self.query_one(DataTable)
        table.add_columns(*COLUMNS)
        self.test()

    @work(thread=True)
    def test(self):
        self.call_from_thread(app.new_device, 54, 600)
        self.call_from_thread(app.new_device, 500, 230)

    def new_device(self, uuid: int, regs: int) -> None:
        table = self.query_one(DataTable)
        cells = (uuid, regs, 0, 0, "N/D", "No")
        table.add_row(*cells)

    def action_toggle_dark(self) -> None:
        """An action to toggle dark mode."""
        self.dark = not self.dark


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

[martibc16]

Hello @williballenthin, I really apreaciate your feedback here!

Your code completely solves my problem with using external threads.

However, I'm faced with the same issue still. I will try to explain my case better:
I have an existing project that I want to add Textual to it. The folder structre (simplified) is as follows:
-- gui.py <- This is the file I'm posting the code from
-- main.py <- This is the existing project that does the heavy lifting in therms of the app.

main.py implements a comunication protocol via USB (which in the end comunicates via BLE with on-the-edge devices). This app is working as expected.

However, we now need a user interface to see what the app (main.py) is doing. Here's where gui.py and the Datatable get into play.

main.py will comunicate with the devices and will extract information from them. This information should be shown in the Datatable. My idea was to send this information to gui.py via methods. So, gui.py would create methods accessible from main.py such as:
new_device()
update_device()
etc.

This way the changed to main.py would be minimal and would consists of calls such as app.new_device or app.update_device etc.

Is this possible? Wjhat would be the better option?

Because from all of your examples the data seems to originate on the same module Textual is running on. In my case the information is generated elsewhere.

[stephenstech-ca]

Hello @willmcgugan and @martibc16.

I'm glad I found this question. I am in the same situation. I'm looking forward to see how to resolve this.

[stephenstech-ca]

@willmcgugan @davep Is there an example app that demonstrates using textual as a TUI frontend for an existing/external app? If not it might be a good candidate for a simple example app as retrofitting existing python code with a new textual frontend TUI might be a common use case.

[drzraf]

I think I understand why the OP is looking for and can be stated like:
How to use my existing asyncio main loop that would consider Textual as its own task(s)?

I already have an existing asyncio.get_event_loop() + loop.create_task() calls + loop.run_forever() doing their job. This could react not to user-input but to network event and calling Textual App update() / update_table() handler on-demand.

I don't want to include existing asyncio application code inside Textual worker management but quite the opposite.
(I've an app doing some aiohttp Client but it seems its internal asyncio code isn't happy with the loop provided by call_from_thread)

Could the app.run() stanza be replaced by myloop.create_task(app.workers.<something TBF>()) that would make Textual plug'n'play which is what I believe OP desired?

[mbastida123]

Exactly. The application doing the task is already done. I just need a pretty picture so that the user can recieve feedback on what the APP is doing under the hood.

[drzraf]

I was able to make it work just adding my core-app as another task of Textual's main asyncio loop:

    def on_mount(self) -> None:
        [...]
        loop = self._loop
        self._loop.create_task(myapp.start(loop))

I passed App._loop down so that underlying components (eg: aiohttp) can rely on it