query_one(...) fails with No nodes match <DOMQuery query='...'>

[smallrobots]

Hello,

I have this HorizontalScroll with three Labels widget as follows:

class JointAngleSpeedDisplay(HorizontalScroll):

    joint_no: reactive[str] = reactive(" Jt 1 ")
    joint_angle: reactive[str] = reactive(" 0.0 ")
    joint_speed: reactive[str] = reactive( " 0.0 ")

    def compose(self) -> ComposeResult:
        yield Label(self.joint_no, id = "no")
        yield Label(self.joint_angle, id = "angle")
        yield Label(self.joint_speed, id = "speed")
        
    def watch_joint_angle(self, angle: float) -> None:
        self.query_one("#angle", Label).update(str(angle))

when I update the reactive joint_angle (outside this class implementation), watch_joint_angle gets called but the DOM query fails saying: No nodes match

What am I missing?

Many thanks,
Riccardo.

[willmcgugan]

Looks like it should work. Are you changing joint_angle before JointAngleSpeedDisplay has been mounted?

Try adding self.log(self.tree) inside watch_join_angle which will log the state of the DOM and that point.

[smallrobots]

Most probably. Where do I retrieve the log?

[willmcgugan]

https://textual.textualize.io/guide/devtools/#console

[davep]

(outside this class implementation)

Normally I've had the error if I've triggered a watch method before the DOM is mounted; is that perhaps the case here?

[smallrobots]

Ok, I too was thinking that the query fails because the DOM is not yet mounted. But then to understand where the issue is, I need to add more context.

The purpose of the tui app is to display live data coming from an external device. To try textual, I'm just generating random numbers. I also would like to point out that, while I have some experience with Threading, I'm not familiar with asyncio (so the problem could be here too).

JointAngleSpeedDisplay is used by ArmControl:

class ArmControl(VerticalScroll):
    def compose(self) -> ComposeResult:
        yield Label("Robot Arm Control", id = "ArmControlTitle")
        yield JointAngleSpeedDisplay(id="j1")

    def update(self, arm_state):
        self.query_one(JointAngleSpeedDisplay).joint_angle = arm_state[0][0]

The main app launches two tasks: arm_state_updater generates the random numbers (it will be modified to connect to the external device), while tui_updater periodically updates the tui.

class MosaicoUI(App):
    def __init__(self):
        super().__init__()
        self.arm_state = array = [[0.0 for j in range(2)] for i in range(6)]  

    def compose(self) -> ComposeResult:
        """Create child widgets for the app."""
        yield Header()
        yield ArmControl()
        yield Footer()
        task1 = asyncio.create_task(self.arm_state_updater())
        task2 = asyncio.create_task(self.tui_updater())

    async def arm_state_updater(self):
        while True:
            for j in range(2):
                for i in range(6):
                    self.arm_state[i][j] = random.uniform(0, 1)
            await asyncio.sleep(0.5)

    async def tui_updater(self):
        while True:
            widget = self.query_one(ArmControl)
            widget.update (self.arm_state)
            await asyncio.sleep(0.5)

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

Maybe the two tasks aren't defined or launched properly?

[willmcgugan]

Your widgets won't be mounted until the compose method has finished. So your tasks risk accessing the DOM before it is ready. A better place to launch tasks would be on_mount.

You might want to use workers which help simplify tasks with Textual apps.