Fast live update of a table

Rich
#1

Hi, I’m just starting out with Textual and Rich so this might be a silly question, but I have been bashing my head against this a few hours and I can’t find a solution in this forum or the API.

I’m doing a simple logger for a CAN-bus; the data arrives quite fast and I want to display it in a table “in real time”.

Here’s the code:

import csv
import io
import can
import binascii
import sys
import time
from rich.console import Console
from rich.table import Table
from rich.live import Live

class Entries:
    entries = { }

    def __init__(self):
        pass

    def addMessage(self, msg):
        can_id = msg.arbitration_id
        dlc = msg.dlc
        ext = int(msg.is_extended_id)
        rtr = int(msg.is_remote_frame)
        data = binascii.hexlify(msg.data).decode("ascii")
        self.entries[can_id] = (can_id, dlc, ext, rtr, data)

    def toCSV(self):
        """
        Convert to CSV
        """
        result = ""
        for i in self.entries.items():
            k = i[1]
            result += "%08X,%d,%d,%d,%s\n" % (k[0],k[1],k[2],k[3],k[4])

        return result

    def generateTable(self) -> Table:
        """
        """
        table = Table()
        table.add_column("ID")
        table.add_column("DlC")
        table.add_column("EXT")
        table.add_column("RTR")
        table.add_column("DATA")
        for i in self.entries.items():
            k = i[1]
            r = "%08X,%d,%d,%d,%s" % (k[0],k[1],k[2],k[3],k[4])
            r = r.split(",")
            table.add_row(r[0],r[1],r[2],r[3],r[4])

        return table

ENTRIES = Entries()
bus = can.interface.Bus(bustype='socketcan', channel='can0', bitrate=500000)
console = Console()

try:
    while 1:
        with Live(ENTRIES.generateTable(), refresh_per_second=10) as live:
            msg = bus.recv(timeout=1.0)
            if msg:
                ENTRIES.addMessage(msg)

            live.update(ENTRIES.generateTable(), refresh=True)

except KeyboardInterrupt:
    bus.shutdown()
    sys.exit(1)

(Yes, its a bit kludgy, but its a work in progress)

This code makes a new table for each update, so the screen is spammed with tables instead of one table being updated. If i do a console.clear() then I get a very non-subtle flickering.

Is it possible to update a table really fast, without getting new tables in the console or flickering?

#2

You can use reactive() for that. Basically, you define global variables (like your entries{}) as reactive.

See the guide on that.

e.g.:

class Entries(Widget):
    entries = reactive({}) #not sure if lists/dictionaries work in reactive, but you get the idea
    ....
    ....

Afterwards, you can change the values of entries like:

...
    table = Entries()
    table.entries = {"some data"}
...

This would also update the table, so instead of generating a new table all the time, you generate one and change the values. You even have the possibility to look (actually watch) for changes on reactive values.

class Entries(Widget):
    entries = reactive({}) #not sure if lists/dictionaries work in reactive, but you get the idea
    ....
    ....
    def watch_entries(self, old_entries, new_entries):
       # do stuff like updates

Hope that helps.