Setup Flow API Reference

The setup flow handles user interaction for device configuration and discovery.

BaseSetupFlow

ucapi_framework.setup.BaseSetupFlow

BaseSetupFlow(config_manager: BaseConfigManager, *, driver: BaseIntegrationDriver, device_class: type | None = None, discovery: BaseDiscovery | None = None, show_migration_in_ui: bool | None = None)

Bases: ABC, Generic[ConfigT]

Base class for integration setup flows.

Handles common patterns: - Configuration mode (add/update/remove/reset) - Device discovery with manual fallback - Device creation and validation - State machine management

Class Type Parameters:

Name	Bound or Constraints	Description	Default
ConfigT		The device configuration class	required

Initialize the setup flow.

Child classes typically don't need to override init - the driver, device_class, and discovery are set automatically by create_handler().

:param config_manager: Device configuration manager instance :param driver: Reference to the driver instance (provides access to driver state) :param device_class: The device class (enables calling class methods for validation) :param discovery: Discovery instance for auto-discovery. Pass None if the device does not support discovery. This is typically instantiated in your driver's main() and passed via create_handler(). :param show_migration_in_ui: Whether to show migration option in configuration mode. Default is None (auto-detect based on get_migration_data override). Set to True/False to explicitly override auto-detection.

Functions

create_handler classmethod

create_handler(driver: BaseIntegrationDriver, discovery: BaseDiscovery | None = None)

Create a setup handler function with the given configuration.

This is a convenience factory method that creates a closure containing the setup flow instance, suitable for passing to IntegrationAPI.init().

The driver_id is automatically extracted from the driver instance. If the driver has a driver_id set, it will be used to automatically fetch the current version from the Remote during migration.

Example usage in driver's main(): discovery = MyDiscovery(api_key="...", timeout=30) setup_handler = MySetupFlow.create_handler(driver, discovery=discovery) api.init("driver-name", setup_handler=setup_handler)

:param driver: The driver instance. The config_manager and driver_id will be retrieved from the driver. :param discovery: Optional initialized discovery instance for auto-discovery. Pass None if the device does not support discovery. :return: Async function that handles SetupDriver messages

handle_driver_setup async

handle_driver_setup(msg: SetupDriver) -> SetupAction

Main dispatcher for setup requests.

:param msg: Setup driver request object :return: Setup action on how to continue

query_device abstractmethod async

query_device(input_values: dict[str, Any]) -> ConfigT | SetupError | RequestUserInput

Query and validate device using collected information.

This method is called after the user provides device information (via manual entry or discovery). This is where you typically have enough info to query the device, validate connectivity, fetch additional data, or perform authentication.

Using Device Class for Validation:

The framework provides self.device_class which you can use to call class methods for validation. This keeps validation logic with your device class:

class MyDevice(StatelessHTTPDevice):
    @classmethod
    async def validate_connection(cls, host: str, token: str) -> dict:
        '''Validate credentials and return device info.'''
        async with aiohttp.ClientSession() as session:
            async with session.get(f"http://{host}/api/info",
                                   headers={"Token": token}) as resp:
                if resp.status != 200:
                    raise ConnectionError("Invalid credentials")
                return await resp.json()

# In your setup flow:
async def query_device(self, input_values):
    try:
        info = await self.device_class.validate_connection(
            host=input_values["host"],
            token=input_values["token"]
        )
        return MyDeviceConfig(
            identifier=info["device_id"],
            name=info["name"],
            host=input_values["host"],
            token=input_values["token"]
        )
    except ConnectionError:
        return SetupError(error_type=IntegrationSetupError.CONNECTION_REFUSED)

Based on the query results, you can: - Return a complete device config to finish setup - Show additional screens to collect more information - Return an error if validation fails

This method can return: - ConfigT: A valid device configuration - if no additional screens needed, setup completes. If you need additional screens, DON'T return the config - store it in self._pending_device_config and return RequestUserInput instead. - SetupError: An error to abort the setup with an error message - RequestUserInput: A screen to display for additional configuration or validation. IMPORTANT: To show additional screens after this one, you MUST set self._pending_device_config BEFORE returning RequestUserInput. The response will then route to handle_additional_configuration_response().

Example - Simple case (no additional screens): async def query_device(self, input_values): # Query the device to validate connectivity device_info = await self.api.get_device_info(input_values["host"])

    if not device_info:
        return SetupError(error_type=IntegrationSetupError.CONNECTION_REFUSED)

    # Just return the config - setup completes automatically
    return MyDeviceConfig(
        identifier=device_info["id"],
        name=input_values["name"],
        address=input_values["host"],
        port=int(input_values.get("port", 8080)),
        version=device_info["version"]
    )

Example - With validation

async def query_device(self, input_values): host = input_values.get("host", "").strip() if not host: return SetupError(error_type=IntegrationSetupError.CONNECTION_REFUSED)

# Test connection
if not await self.api.test_connection(host):
    return SetupError(error_type=IntegrationSetupError.CONNECTION_REFUSED)

return MyDeviceConfig(
    identifier=host,
    name=input_values.get("name", host),
    address=host
)

Example - Multi-screen flow (query device, then show additional options): async def query_device(self, input_values): # Query the device API to validate and fetch available options auth_response = await self.api.authenticate( input_values["host"], input_values["token"] )

    if not auth_response["valid"]:
        return SetupError(error_type=IntegrationSetupError.AUTHORIZATION_ERROR)

    # IMPORTANT: Store config in _pending_device_config for multi-screen flows
    self._pending_device_config = MyDeviceConfig(
        identifier=input_values["host"],
        name=input_values["name"],
        token=auth_response["token"],
        available_servers=auth_response["servers"]  # Data needed for next screen
    )

    # Return screen - response will route to handle_additional_configuration_response
    return RequestUserInput(
        {"en": "Select Server"},
        [{"id": "server", "label": {"en": "Server"},
          "field": {"dropdown": {"items": self._build_server_dropdown()}}}]
    )

async def handle_additional_configuration_response(self, msg):
    # Access stored config and new input
    self._pending_device_config.server = msg.input_values["server"]
    return None  # Save and complete (or return modified config)

Example - Re-display form with validation error

async def query_device(self, input_values): host = input_values.get("host", "").strip() if not host: # Show the form again with error (no _pending_device_config set) return RequestUserInput( {"en": "Invalid Input"}, [ {"id": "error", "label": {"en": "Error"}, "field": {"label": {"value": {"en": "Host is required"}}}}, # ... rest of the form fields ] )

return MyDeviceConfig(identifier=host, name=host, address=host)

:param input_values: User input values from the manual entry form. Also includes self._pre_discovery_data if pre-discovery screens were shown. :return: Device configuration, SetupError, or RequestUserInput to re-display form

get_manual_entry_form abstractmethod

get_manual_entry_form() -> RequestUserInput

Get the manual entry form.

:return: RequestUserInput with manual entry fields

discover_devices async

discover_devices() -> list[DiscoveredDevice]

Perform device discovery.

DEFAULT IMPLEMENTATION: Calls self.discovery.discover() if available.

If a discovery_class was passed to init, this method will call its discover() method and return the results. If no discovery_class was provided (None), this returns an empty list and the setup flow will skip discovery.

:return: List of discovered devices, or empty list if discovery not supported

prepare_input_from_discovery async

prepare_input_from_discovery(discovered: DiscoveredDevice, additional_input: dict[str, Any]) -> dict[str, Any]

Convert discovered device data to input_values format for query_device.

This method transforms a discovered device into the same input_values format that manual entry produces. This allows query_device() to work uniformly for both discovery and manual entry paths.

The returned dictionary should match the field names from your manual entry form, so query_device() can process both sources identically.

DEFAULT IMPLEMENTATION: Returns a basic dictionary with common fields. Override this to customize the mapping for your integration.

:param discovered: The discovered device selected by the user :param additional_input: Additional user input from the discovery screen (e.g., from get_additional_discovery_fields) :return: Dictionary of input values in the same format as manual entry

Example - Basic mapping

async def prepare_input_from_discovery(self, discovered, additional_input): return { "identifier": discovered.identifier, "address": discovered.address, "name": discovered.name, "port": discovered.extra_data.get("port", 8080), # Include any additional fields from discovery screen **additional_input }

Example - With data transformation

async def prepare_input_from_discovery(self, discovered, additional_input): # Extract specific data from extra_data return { "identifier": discovered.identifier, "address": discovered.address, "name": additional_input.get("name", discovered.name), # Allow override "model": discovered.extra_data.get("model"), "firmware": discovered.extra_data.get("version"), }

Example - With filtering

async def prepare_input_from_discovery(self, discovered, additional_input): # Only include relevant additional input fields return { "identifier": discovered.identifier, "address": discovered.address, "name": discovered.name, # Only include specific additional fields, not "choice" "zone": additional_input.get("zone", 1), "volume_step": additional_input.get("volume_step", 5), }

get_discovered_devices

get_discovered_devices(identifier: str | None = None) -> list[DiscoveredDevice] | DiscoveredDevice | None

Get discovered devices from the last discovery run.

This is a convenience method that returns devices found by the framework's automatic discovery. Use this in your create_device_from_discovery() implementation to access device details.

This is equivalent to accessing self.discovery.devices directly.

:param identifier: Optional device identifier to look up a specific device. If provided, returns the matching DiscoveredDevice or None. If omitted, returns the full list of devices. :return: If identifier provided: DiscoveredDevice or None If no identifier: List of all discovered devices (empty if none found)

Example - Get specific device

async def create_device_from_discovery(self, device_id, additional_data): discovered = self.get_discovered_devices(device_id) if not discovered: return SetupError(error_type=IntegrationSetupError.NOT_FOUND)

return MyDeviceConfig(
    identifier=discovered.identifier,
    name=discovered.name,
    address=discovered.address,
    port=discovered.extra_data.get("port", 80)
)

Example - Get all devices

async def create_device_from_discovery(self, device_id, additional_data): for device in self.get_discovered_devices(): if device.identifier == device_id: return MyDeviceConfig.from_discovered(device) return SetupError(error_type=IntegrationSetupError.NOT_FOUND)

get_device_id

get_device_id(device_config: ConfigT) -> str

Extract device ID from configuration.

Default implementation: tries common attribute names (identifier, id, device_id). Override this if your config uses a different attribute name.

:param device_config: Device configuration :return: Device identifier :raises AttributeError: If no valid ID attribute is found

get_device_name

get_device_name(device_config: ConfigT) -> str

Extract device name from configuration.

Default implementation: tries common attribute names (name, friendly_name, device_name). Override this if your config uses a different attribute name.

:param device_config: Device configuration :return: Device name :raises AttributeError: If no valid name attribute is found

format_discovered_device_label

format_discovered_device_label(device: DiscoveredDevice) -> str

Format how a discovered device appears in the dropdown list.

Override this method to customize how devices are displayed to users during discovery. The default format shows the device name and address.

:param device: The discovered device to format :return: Formatted label string

Example - Include model information

def format_discovered_device_label(self, device): model = device.extra_data.get("model", "Unknown") return f"{device.name} - {model} ({device.address})"

Example - Show additional details

def format_discovered_device_label(self, device): version = device.extra_data.get("version", "") return f"{device.name} [{version}] at {device.address}"

get_discovered_devices_screen async

get_discovered_devices_screen(devices: list[DiscoveredDevice]) -> RequestUserInput

Build the discovered devices selection screen.

Override this method to completely customize the discovery screen layout, such as adding additional fields, changing the title, or using a different input type.

The default implementation creates a dropdown with all discovered devices (using format_discovered_device_label for labels), plus a "Setup Manually" option, and includes any additional fields from get_additional_discovery_fields().

The selected device's identifier will be passed to create_device_from_discovery().

:param devices: List of discovered devices :return: RequestUserInput screen to show to the user

Example - Custom screen with additional fields

async def get_discovered_devices_screen(self, devices): dropdown_items = [ { "id": d.identifier, "label": {"en": self.format_discovered_device_label(d)} } for d in devices ] dropdown_items.append({"id": "manual", "label": {"en": "Manual Setup"}})

return RequestUserInput(
    {"en": "Select Your Device"},
    [
        {
            "id": "choice",
            "label": {"en": "Available Devices"},
            "field": {"dropdown": {"value": dropdown_items[0]["id"], "items": dropdown_items}}
        },
        {
            "id": "zone",
            "label": {"en": "Default Zone"},
            "field": {"number": {"value": 1, "min": 1, "max": 10}}
        }
    ]
)

get_additional_discovery_fields

get_additional_discovery_fields() -> list[dict]

Get additional fields to show during discovery.

Override to add custom fields (e.g., volume step, zone selection).

:return: List of field definitions

extract_additional_setup_data

extract_additional_setup_data(input_values: dict[str, Any]) -> dict[str, Any]

Extract additional setup data from input values.

Override to extract additional custom fields.

:param input_values: User input values :return: Dictionary of additional data

get_pre_discovery_screen async

get_pre_discovery_screen() -> RequestUserInput | None

Request pre-discovery configuration screen(s).

Override this method to show configuration screens BEFORE device discovery. This is useful for collecting credentials, API keys, server addresses, or other information needed to perform discovery.

The collected data is stored in self._pre_discovery_data and can be accessed during discovery (in discover_devices()) or device creation.

To show a pre-discovery screen: 1. Return a RequestUserInput with the fields you need 2. Handle the response in handle_pre_discovery_response() 3. Return another RequestUserInput to show more screens, or None to proceed

:return: RequestUserInput to show a screen, or None to skip pre-discovery

handle_pre_discovery_response async

handle_pre_discovery_response(msg: UserDataResponse) -> SetupAction | None

Handle response from pre-discovery screens.

Override this method to process responses from screens created by get_pre_discovery_screen(). The input values are automatically stored in self._pre_discovery_data before this method is called.

You should: 1. Validate the input (optionally) 2. Either: - Return another RequestUserInput for more pre-discovery screens, or - Return None to proceed to device discovery

If you return None, the base class will call discover_devices() where you can access self._pre_discovery_data to use the collected information.

:param msg: User data response from pre-discovery screen :return: RequestUserInput for another screen, or None to proceed to discovery

get_additional_configuration_screen async

get_additional_configuration_screen(device_config: ConfigT, previous_input: dict[str, Any]) -> RequestUserInput | None

Request additional configuration screens after device creation.

Override this method to show additional setup screens that collect more information about the device. This is called after query_device (for both manual entry and discovery paths) but BEFORE the device is saved.

AUTO-POPULATION: Any fields returned by this screen will automatically populate matching attributes on self._pending_device_config. You typically don't need to manually handle the response!

Example - Simple additional screen

async def get_additional_configuration_screen(self, device_config, previous_input): return RequestUserInput( {"en": "Additional Settings"}, [ {"id": "token", "label": {"en": "API Token"}, "field": {"text": {"value": ""}}}, {"id": "zone", "label": {"en": "Zone"}, "field": {"number": {"value": 1}}} ] ) # token and zone will auto-populate if device_config has those attributes!

Example - Conditional screen

async def get_additional_configuration_screen(self, device_config, previous_input): if device_config.requires_auth: return RequestUserInput( {"en": "Authentication"}, [{"id": "password", "label": {"en": "Password"}, "field": {"text": {"value": ""}}}] ) return None # No additional screen needed

:param device_config: The device configuration (also in self._pending_device_config) :param previous_input: Input values from the previous screen :return: RequestUserInput to show another screen, or None to complete setup

handle_additional_configuration_response async

handle_additional_configuration_response(msg: UserDataResponse) -> ConfigT | SetupAction | None

Handle response from additional configuration screens.

Override this method to process responses from custom setup screens created by get_additional_configuration_screen().

AUTO-POPULATION: The framework automatically populates self._pending_device_config from msg.input_values where field names match config attributes. In most cases, you don't need to override this method at all!

Return one of: - None (recommended): Auto-populated fields are saved automatically - ConfigT (device config): Replace pending config and save this one - RequestUserInput: Show another configuration screen - SetupError: Abort setup with an error

Example - No override needed (auto-population): # If your screen has fields like "token" and "zone" that match # attributes on your device config, they're automatically set! # No need to override handle_additional_configuration_response at all.

Example - With validation

async def handle_additional_configuration_response(self, msg): # Fields already auto-populated, just validate if not self._pending_device_config.token: return SetupError(error_type=IntegrationSetupError.AUTHORIZATION_ERROR)

# Or add computed fields
self._pending_device_config.full_url = (
    f"https://{self._pending_device_config.address}:8080"
)
return None  # Save and complete

Example - Show another screen

async def handle_additional_configuration_response(self, msg): # Check if we need authentication if self._pending_device_config.requires_auth: return RequestUserInput( {"en": "Enter Password"}, [{"id": "password", "label": {"en": "Password"}, "field": {"text": {"value": ""}}}] ) return None

Example - Replace entire config (advanced): async def handle_additional_configuration_response(self, msg): # Create completely new config (rarely needed) return MyDeviceConfig( identifier=self._pending_device_config.identifier, name=self._pending_device_config.name, address=self._pending_device_config.address, token=msg.input_values["token"], # Manual access if needed )

:param msg: User data response from additional screen :return: Device config to save, SetupAction, or None to complete

SetupSteps

ucapi_framework.setup.SetupSteps

Bases: IntEnum

Enumeration of setup steps to keep track of user data responses.