UIManager

The UIManager class is the central controller for the pydoover UI system. It manages UI state, handles communication with the cloud, coordinates updates, and processes incoming commands.

Import

from pydoover.ui import UIManager

Constructor

UIManager(
    app_key: str = None,
    client: Union[Client, DeviceAgentInterface] = None,
    auto_start: bool = False,
    min_ui_update_period: int = 600,
    min_observed_update_period: int = 4,
    is_async: bool = None,
)

Parameters

Parameter	Type	Default	Description
`app_key`	`str`	`None`	Unique identifier for your application, used to namespace UI elements
`client`	`Client` or `DeviceAgentInterface`	`None`	The client connection for cloud communication
`auto_start`	`bool`	`False`	If `True`, automatically starts communication setup
`min_ui_update_period`	`int`	`600`	Minimum seconds between UI updates when not observed
`min_observed_update_period`	`int`	`4`	Minimum seconds between UI updates when observed by users
`is_async`	`bool`	`None`	Force async or sync mode; auto-detected if `None`

Basic Usage

from pydoover.ui import UIManager, NumericVariable

# Create the manager
ui = UIManager(app_key="my_app", client=dda_interface)

# Add UI elements
temp = NumericVariable("temperature", "Temperature", curr_val=25.0)
ui.add_children(temp)

# Start communication
ui.start_comms()

# Update and push
ui.update_variable("temperature", 26.5)
ui.push()

Connection Management

start_comms()

Initializes subscriptions for UI state and command channels.

ui.start_comms()

is_connected()

Checks if the UI manager has an active connection to the cloud.

if ui.is_connected():
    print("Connected to cloud")

has_been_connected()

Checks if the connection has ever been established (useful for startup sequences).

if ui.has_been_connected():
    print("Connection has been established at some point")

is_being_observed()

Returns True if users are currently viewing the UI. This affects update frequency.

if ui.is_being_observed():
    # More frequent updates are appropriate
    pass

await_comms_sync()

Async method to wait for channel synchronization.

synced = await ui.await_comms_sync(timeout=5)
if synced:
    print("Channels synchronized")

Managing UI Elements

add_children()

Adds one or more UI elements to the root container.

from pydoover.ui import NumericVariable, TextVariable, Submodule

# Add individual elements
ui.add_children(
    NumericVariable("temp", "Temperature", curr_val=25.0),
    TextVariable("status", "Status", curr_val="Running")
)

# Add a submodule with children
module = Submodule("sensors", "Sensors", children=[
    NumericVariable("humidity", "Humidity", curr_val=60.0)
])
ui.add_children(module)

remove_children()

Removes elements from the UI.

ui.remove_children(element_to_remove)

set_children()

Replaces all children with a new list.

ui.set_children([new_element1, new_element2])

get_element()

Retrieves an element by name from anywhere in the UI hierarchy.

temp_element = ui.get_element("temperature")
if temp_element:
    print(f"Current value: {temp_element.current_value}")

get_all_variables()

Returns a list of all Variable elements.

for var in ui.get_all_variables():
    print(f"{var.name}: {var.current_value}")

Managing Interactions

add_interaction()

Adds an interactive element (command) to the UI.

from pydoover.ui import Action, Colour

restart = Action(
    name="restart",
    display_name="Restart",
    colour=Colour.red,
    requires_confirm=True
)
ui.add_interaction(restart)

get_interaction()

Retrieves an interaction by name.

restart_cmd = ui.get_interaction("restart")

get_all_interactions()

Returns a list of all registered interactions.

for interaction in ui.get_all_interactions():
    print(f"Interaction: {interaction.name}")

get_all_interaction_names()

Returns a list of all interaction names.

names = ui.get_all_interaction_names()

coerce_command()

Programmatically sets an interaction's value (useful for setting defaults or resetting state).

ui.coerce_command("my_slider", 50, critical=True)

Updating Variables

update_variable()

Updates a variable's value by name.

# Simple update
ui.update_variable("temperature", 27.5)

# Critical update (triggers immediate sync)
ui.update_variable("alarm_status", True, critical=True)

Synchronization Methods

pull() / pull_async()

Fetches the current UI state and commands from the cloud.

# Synchronous
ui.pull()

# Asynchronous
await ui.pull_async()

push() / push_async()

Sends local UI state and command changes to the cloud.

# Basic push
ui.push()

# With options
ui.push(
    record_log=True,      # Log this update
    max_age=60,           # Cache duration hint
    should_remove=True,   # Remove elements not in local state
    even_if_empty=False,  # Don't push if no changes
)

# Asynchronous
await ui.push_async()

handle_comms() / handle_comms_async()

Intelligently manages synchronization based on observation state and update thresholds.

# Recommended approach for periodic updates
ui.handle_comms()

# With forced logging
ui.handle_comms(force_log=True)

# Asynchronous
await ui.handle_comms_async()

clear_ui() / clear_ui_async()

Clears all UI state from the cloud.

ui.clear_ui()

# Async
await ui.clear_ui_async()

Callbacks and Event Handling

Registering Callbacks

Use the @callback decorator to register handlers for interaction changes.

import re

# Single element callback
@ui.callback("my_button")
def on_button_press(element, new_value):
    print(f"Button pressed: {new_value}")

# Pattern-matching callback
@ui.callback(re.compile(r"sensor_\d+_toggle"))
def on_sensor_toggle(element, new_value):
    print(f"{element.name} toggled to {new_value}")

# Global interaction callback
@ui.callback("camera_get_now", global_interaction=True)
def on_camera_capture(element, new_value):
    print("Camera capture requested")

register_callbacks()

Registers all callback-decorated methods from an object.

class MyController:
    @callback("start_button")
    def on_start(self, element, new_value):
        pass

controller = MyController()
ui.register_callbacks(controller)

register_interactions()

Registers all interaction-decorated methods from an object.

class MyController:
    @action("restart", "Restart Device")
    def restart_device(self, new_value):
        pass

controller = MyController()
ui.register_interactions(controller)

add_cmds_update_subscription()

Adds a callback that fires on any command update.

def on_any_command():
    print("Commands were updated")

ui.add_cmds_update_subscription(on_any_command)

Application Display Settings

set_display_name()

Sets the display name for the application.

ui.set_display_name("My IoT Device")

set_status_icon()

Sets a status icon for the application.

ui.set_status_icon("warning")  # Or "ok", "error", etc.

set_variant()

Sets how the application is displayed.

from pydoover.ui import ApplicationVariant

ui.set_variant(ApplicationVariant.stacked)   # Elements stacked
ui.set_variant(ApplicationVariant.submodule) # Wrapped in submodule

set_position()

Sets the display position order for the application.

ui.set_position(10)  # Lower numbers appear first

Notifications

send_notification() / send_notification_async()

Sends a notification to users.

ui.send_notification("Temperature threshold exceeded!")

# Without activity log
ui.send_notification("Info message", record_activity=False)

# Async
await ui.send_notification_async("Alert!")

Publishing to Channels

publish_to_channel()

Publishes data to a named channel.

ui.publish_to_channel(
    "custom_data",
    {"sensor_reading": 42.5, "timestamp": time.time()},
    record_log=True,
    max_age=60
)

Properties and State

Property	Description
`app_key`	The application key
`client`	The connection client
`last_ui_state`	Dictionary of the last known UI state from cloud
`last_ui_cmds`	Dictionary of the last known commands from cloud

Complete Example

from pydoover.ui import (
    UIManager,
    NumericVariable,
    TextVariable,
    Action,
    Submodule,
    Colour,
    Range,
    action,
    callback,
)

class DeviceController:
    def __init__(self, dda_interface):
        self.ui = UIManager(
            app_key="device_controller",
            client=dda_interface,
            min_ui_update_period=300,
            min_observed_update_period=2,
        )
        self.setup_ui()
        self.ui.register_interactions(self)
        self.ui.register_callbacks(self)
        self.ui.start_comms()

    def setup_ui(self):
        # Create variables
        self.temperature = NumericVariable(
            name="temperature",
            display_name="Temperature",
            curr_val=25.0,
            precision=1,
            ranges=[
                Range("Normal", 15, 30, Colour.green),
                Range("High", 30, 45, Colour.red),
            ]
        )

        self.status = TextVariable(
            name="status",
            display_name="Status",
            curr_val="Initializing"
        )

        # Create submodule
        sensors = Submodule(
            name="sensors",
            display_name="Sensors",
            children=[self.temperature, self.status]
        )

        self.ui.add_children(sensors)

    @action("restart", "Restart Device", colour=Colour.red, requires_confirm=True)
    def restart(self, new_value):
        print("Restarting device...")
        self.status.current_value = "Restarting"

    @callback("temperature")
    def on_temp_change(self, element, new_value):
        if new_value > 35:
            self.ui.send_notification(f"High temperature: {new_value}")

    async def update_loop(self):
        while True:
            # Update sensor values
            self.temperature.current_value = read_temperature()
            self.status.current_value = "Running"

            # Handle communication
            await self.ui.handle_comms_async()

            await asyncio.sleep(1)