Application Class

from pydoover.docker import Application, run_app

class Application:
    def __init__(
        self,
        config: Schema,
        app_key: str = None,
        is_async: bool = None,
        device_agent: DeviceAgentInterface = None,
        platform_iface: PlatformInterface = None,
        modbus_iface: ModbusInterface = None,
        name: str = None,
        test_mode: bool = False,
        config_fp: str = None,
        healthcheck_port: int = None,
    )

def setup(self):
    """Initialize resources and set up the application."""
    pass

# Or async version
async def setup(self):
    """Async initialization."""
    await self.async_init()

def main_loop(self):
    """Main application logic."""
    pass

# Or async version
async def main_loop(self):
    """Async main loop."""
    data = await self.fetch_data()

def get_tag(
    self,
    tag_key: str,
    app_key: str = None,
    default: Any = None
) -> Any | None

# Get tag from current app
temperature = self.get_tag("temperature")

# Get tag with default
status = self.get_tag("status", default="unknown")

# Get tag from another app
other_value = self.get_tag("sensor_reading", app_key="other-app-1234")

def get_global_tag(self, tag_key: str, default: Any = None) -> Any | None

system_status = self.get_global_tag("system_status")
shutdown_requested = self.get_global_tag("shutdown_requested", default=False)

def set_tag(
    self,
    tag_key: str,
    value: Any,
    app_key: str = None,
    only_if_changed: bool = True
) -> None

# Set a tag for current app
self.set_tag("temperature", 25.5)

# Set a tag for another app
self.set_tag("command", "start", app_key="other-app-1234")

# Always publish even if unchanged
self.set_tag("heartbeat", True, only_if_changed=False)

await self.set_tag_async("temperature", 25.5)

def set_tags(
    self,
    tags: dict[str, Any],
    app_key: str = None,
    only_if_changed: bool = True
) -> None

self.set_tags({
    "temperature": 25.5,
    "humidity": 60.0,
    "status": "online"
})

def set_global_tag(
    self,
    tag_key: str,
    value: Any,
    only_if_changed: bool = True
) -> None

self.set_global_tag("system_status", "operational")

def subscribe_to_tag(
    self,
    tag_key: str,
    callback: Callable[[str, Any], Any],
    app_key: str = None,
    global_tag: bool = False
)

def on_temperature_change(tag_key, new_value):
    print(f"Temperature changed to {new_value}")

self.subscribe_to_tag("temperature", on_temperature_change)

def get_di(self, di: int) -> bool

button_pressed = self.get_di(0)

def get_ai(self, ai: int) -> float

voltage = self.get_ai(0)

def get_do(self, do: int) -> bool

relay_state = self.get_do(0)

def set_do(self, do: int, value: bool) -> None

self.set_do(0, True)  # Turn on relay

def schedule_do(self, do: int, value: bool, delay_secs: int) -> None

self.schedule_do(0, False, 10)  # Turn off in 10 seconds

def get_ao(self, ao: int) -> float

def set_ao(self, ao: int, value: float) -> None

self.set_ao(0, 3.3)  # Set to 3.3V

def schedule_ao(self, ao: int, value: float, delay_secs: int) -> None

def subscribe_to_channel(
    self,
    channel_name: str,
    callback: Callable[[str, dict], Any]
)

async def on_config_update(channel_name, data):
    print(f"Config updated: {data}")

self.subscribe_to_channel("deployment_config", on_config_update)

def publish_to_channel(self, channel_name: str, data: str | dict)

# Publish a dictionary
self.publish_to_channel("sensor_data", {"temperature": 25.5})

# Publish a string (useful for alerts and notifications)
self.publish_to_channel("significantAlerts", "Temperature threshold exceeded!")

def get_channel_aggregate(self, channel_name: str) -> dict | None

def set_ui_elements(self, elements: list)

def get_command(self, name: str) -> Any

def coerce_command(self, name: str, value: Any)

def set_ui_status_icon(self, icon: str)

def read_modbus_registers(
    self,
    address: int,
    count: int,
    register_type: int,
    modbus_id: int = None,
    bus_id: str = None
) -> int | list[int] | None

def write_modbus_registers(
    self,
    address: int,
    values: list[int],
    register_type: int,
    modbus_id: int = None,
    bus_id: str = None
) -> bool

def add_new_modbus_read_subscription(
    self,
    address: int,
    count: int,
    register_type: int,
    callback: Callable,
    poll_secs: int = None,
    modbus_id: int = None,
    bus_id: str = None
)

def request_shutdown(self) -> None

async def on_shutdown_at(self, dt: datetime) -> None:
    """Called when shutdown is scheduled."""
    print(f"Shutdown at {dt}")

async def check_can_shutdown(self) -> bool:
    """Return True if safe to shutdown."""
    return True

@property
def is_ready(self) -> bool

await app.wait_until_ready()

def get_is_dda_available(self) -> bool

def get_is_dda_online(self) -> bool

def get_has_dda_been_online(self) -> bool

async def test_my_app():
    app = MyApp(config=MyConfig(), test_mode=True)
    asyncio.create_task(run_app(app, start=False))

    # Wait for app to start
    await app.wait_until_ready()

    # Run one iteration of the main loop
    await app.next()

    # Check state
    assert app.get_tag("status") == "ready"

await app.next()

def run_app(
    app: Application,
    start: bool = True,
    setup_logging: bool = True,
    log_formatter: logging.Formatter = None,
    log_filters: logging.Filter | list[logging.Filter] = None
)

from pydoover.docker import Application, run_app

class MyApp(Application):
    def setup(self):
        pass

    def main_loop(self):
        pass

if __name__ == "__main__":
    run_app(MyApp(config=MyConfig()))

import time
from pydoover.docker import Application, run_app
from pydoover import ui

# app_config.py - Configuration schema
from pydoover import config

class SampleConfig(config.Schema):
    def __init__(self):
        self.outputs_enabled = config.Boolean("Digital Outputs Enabled", default=True)
        self.funny_message = config.String("A Funny Message")
        self.sim_app_key = config.Application("Simulator App Key")

# app_ui.py - UI elements
class SampleUI:
    def __init__(self):
        self.is_working = ui.BooleanVariable("is_working", "We Working?")
        self.uptime = ui.DateTimeVariable("uptime", "Started")
        self.send_alert = ui.Action("send_alert", "Send message as alert")
        self.test_message = ui.TextParameter("test_message", "Put in a message")
        self.test_output = ui.TextVariable("test_output", "Message received")

    def fetch(self):
        return [self.is_working, self.uptime, self.send_alert,
                self.test_message, self.test_output]

# application.py - Main application
class SampleApplication(Application):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.started = time.time()
        self.ui: SampleUI = None

    async def setup(self):
        self.ui = SampleUI()
        self.ui_manager.add_children(*self.ui.fetch())

    async def main_loop(self):
        self.ui.is_working.update(True)
        self.ui.uptime.update(self.started)

        # Read a tag from another application
        random_value = self.get_tag("random_value", self.config.sim_app_key.value)

    @ui.callback("send_alert")
    async def on_send_alert(self, new_value):
        await self.publish_to_channel("significantAlerts", self.ui.test_output.current_value)
        self.ui.send_alert.coerce(None)

    @ui.callback("test_message")
    async def on_text_parameter_change(self, new_value):
        self.ui.test_output.update(new_value)

# Entry point
def main():
    run_app(SampleApplication(config=SampleConfig()))