Add ability to generate GUI forms from pydantic models

.github/workflows/ci.yml

@@ -23,4 +23,4 @@ jobs:
     with:
         python-version-path: .python-version
         runs-on: windows-latest
-        run-coverage: true
+        run-coverage: true

pyproject.toml

@@ -40,11 +40,17 @@ aind-services = [
     "aind-data-schema<2"
 ]
 
+magicgui = [
+    "magicgui[pyqt6]"
+]
+
 dev = [
     "ruff",
     "codespell",
     "coverage",
     "aind-behavior-experiment-launcher[aind-services]",
+    "aind-behavior-experiment-launcher[magicgui]",
+
 ]
 
 docs = [

src/aind_behavior_experiment_launcher/data_transfer/aind_watchdog.py

@@ -1,12 +1,12 @@
-try:
-    import aind_watchdog_service  # noqa: F401
-except ImportError as e:
-    e.add_note(
-        "The 'aind-watchdog-service' package is required to use this module. \
+import importlib.util
+
+if importlib.util.find_spec("aind_watchdog_service") is None:
+    raise ImportError(
+        "The 'aind_watchdog_service' package is required to use this module. \
             Install the optional dependencies defined in `project.toml' \
                 by running `pip install .[aind-services]`"
     )
-    raise
+
 
 import datetime
 import json

src/aind_behavior_experiment_launcher/launcher/behavior_launcher.py

@@ -521,7 +521,7 @@ def pick_rig(self) -> TRig:
         else:
             while True:
                 try:
-                    path = self.prompt_pick_file_from_list(available_rigs, prompt="Choose a rig:", zero_label=None)
+                    path = self.ui_helper.prompt_pick_from_list(available_rigs, prompt="Choose a rig:")
                     if not isinstance(path, str):
                         raise ValueError("Invalid choice.")
                     rig = model_from_json_file(path, self.launcher.rig_schema_model)
@@ -606,7 +606,7 @@ def pick_task_logic(self) -> TTaskLogic:
                 available_files = glob.glob(os.path.join(_path, "*.json"))
                 if len(available_files) == 0:
                     break
-                path = self.prompt_pick_file_from_list(available_files, prompt="Choose a task logic:", zero_label=None)
+                path = self.ui_helper.prompt_pick_from_list(available_files, prompt="Choose a task logic:")
                 if not isinstance(path, str):
                     raise ValueError("Invalid choice.")
                 if not os.path.isfile(path):
@@ -622,51 +622,6 @@ def pick_task_logic(self) -> TTaskLogic:
             raise ValueError("No task logic file found.")
         return task_logic
 
-    def prompt_pick_file_from_list(
-        self,
-        available_files: list[str],
-        prompt: str = "Choose a file:",
-        zero_label: Optional[str] = None,
-        zero_value: Optional[_T] = None,
-        zero_as_input: bool = True,
-        zero_as_input_label: str = "Enter manually",
-    ) -> Optional[str | _T]:
-        """
-        Prompts the user to pick a file from a list of available files.
-
-        Args:
-            available_files (list[str]): List of file paths to choose from.
-            prompt (str): The prompt message to display.
-            zero_label (Optional[str]): Label for the "zero" option.
-            zero_value (Optional[_T]): Value to return for the "zero" option.
-            zero_as_input (bool): Whether to allow manual input for the "zero" option.
-            zero_as_input_label (str): Label for manual input prompt.
-
-        Returns:
-            Optional[str | _T]: The selected file path or the zero value.
-
-        Raises:
-            ValueError: If an invalid choice is made.
-        """
-        self.ui_helper.print(prompt)
-        if zero_label is not None:
-            self.ui_helper.print(f"0: {zero_label}")
-        for i, file in enumerate(available_files):
-            self.ui_helper.print(f"{i + 1}: {os.path.split(file)[1]}")
-        choice = int(input("Choice: "))
-        if choice < 0 or choice >= len(available_files) + 1:
-            raise ValueError
-        if choice == 0:
-            if zero_label is None:
-                raise ValueError
-            else:
-                if zero_as_input:
-                    return str(input(zero_as_input_label))
-                else:
-                    return zero_value
-        else:
-            return available_files[choice - 1]
-
     def choose_subject(self, directory: str | os.PathLike) -> str:
         """
         Prompts the user to select or manually enter a subject name.

src/aind_behavior_experiment_launcher/ui/_magicgui.py

@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+import importlib.util
+
+if importlib.util.find_spec("magicgui") is None:
+    raise ImportError(
+        "The 'magicgui' package is required to use this module. "
+        "Install the optional dependencies defined in `project.toml` "
+        "by running `pip install .[magicgui]`"
+    )
+
+
+from functools import partial
+from typing import Any, Iterable, Optional, Type
+
+from magicgui import event_loop
+from magicgui.types import Undefined
+from magicgui.widgets import (
+    Button,
+    CheckBox,
+    Container,
+    Textarea,
+    create_widget,
+)
+from magicgui.widgets.bases import ValueWidget
+from pydantic import BaseModel, TypeAdapter, ValidationError
+from pydantic.fields import FieldInfo, PydanticUndefined
+
+
+class _ValidationErrorSink:
+    def __init__(self, widget: Optional[ValueWidget] = None):
+        self._widget = widget
+        self._error: Optional[Exception] = None
+        self.clear()
+
+    def __call__(self, e: Exception | None):
+        self.update(e)
+
+    @property
+    def widget(self) -> Optional[ValueWidget]:
+        return self._widget
+
+    @property
+    def error(self) -> Optional[Exception]:
+        return self._error
+
+    def update(self, e: Exception | None):
+        self._error = e
+        if self._widget is not None:
+            self._widget.value = f"{self.error}"
+
+    def clear(self):
+        self.update(None)
+
+    def register_widget(self, widget: ValueWidget):
+        self._widget = widget
+
+
+def _container_updater(container, validator, error_sink: Optional[_ValidationErrorSink] = None):
+    # https://icon-sets.iconify.design/material-symbols/
+
+    null_checkbox: CheckBox = container[0]
+    input_widget: ValueWidget = container[1]
+    validation_button: Button = container[2]
+    if error_sink is None:
+        error_sink = _ValidationErrorSink()
+
+    is_null = null_checkbox.value
+    with input_widget.changed.blocked():
+        input_widget.enabled = not is_null
+        try:
+            if is_null:
+                value = None
+            else:
+                value = input_widget.value
+
+            value = validator(value)
+            if value is not None:
+                # in case the type adapter does something fancy
+                input_widget.set_value(value)
+
+            validation_button.set_icon("material-symbols:check-circle-outline-rounded", "#0b7500")
+            error_sink.clear()
+        except ValidationError as e:
+            validation_button.set_icon("material-symbols:cancel", "#fd0000")
+            error_sink(e)
+
+
+def _make_widget_from_field(
+    field: FieldInfo,
+    name: str,
+    default: Any = Undefined,
+    *,
+    override_type: Optional[dict[str, Type[ValueWidget]]] = None,
+    validation_error_sink: Optional[_ValidationErrorSink] = None,
+) -> Container:
+    # Each field will be rendered as:
+    # - Left button for nullability
+    # - Value input
+    # - Right button for validation
+
+    if override_type is None:
+        override_type = {}
+    widget_type = override_type.get(name, None)
+
+    if default is Undefined:
+        default = field.get_default(call_default_factory=True)
+        if default is PydanticUndefined or default is None:
+            default = Undefined
+
+    container: Container = Container(layout="horizontal")
+
+    container.append(
+        CheckBox(
+            value=(default is Undefined),
+            label="N",
+        )
+    )
+    widget = create_widget(default, field.annotation, name=name, widget_type=widget_type)
+
+    container.append(widget)
+    container.append(Button(icon="material-symbols:check-circle-outline-rounded", icon_color="#0b7500", enabled=True))
+    container.tooltip = field.description
+    validator = TypeAdapter(field.rebuild_annotation()).validate_python
+    updater = partial(_container_updater, validator=validator, error_sink=validation_error_sink)
+    container.changed.connect(updater)
+    updater(container)
+    return container
+
+
+def create_container_from_model(
+    model: Type[BaseModel] | BaseModel,
+    *,
+    include_fields: Optional[Iterable[str]] = None,
+    exclude_fields: Optional[Iterable[str]] = None,
+    override_type: Optional[dict[str, Type[ValueWidget]]] = None,
+    validation_error_sink: Optional[_ValidationErrorSink] = None,
+    populate_with_instance: bool = False,
+) -> "Container":
+    if include_fields is None:
+        include_fields = model.model_fields.keys()
+    if exclude_fields is None:
+        exclude_fields = []
+
+    if populate_with_instance:
+        if not isinstance(model, BaseModel):
+            raise ValueError("Cannot populate with instance if model is not an instance")
+
+    widgets = [
+        _make_widget_from_field(
+            field,
+            name,
+            getattr(model, name) if populate_with_instance else Undefined,
+            override_type=override_type,
+            validation_error_sink=validation_error_sink,
+        )
+        for name, field in model.model_fields.items()
+        if (name in include_fields) and (name not in exclude_fields)
+    ]
+
+    container = Container(widgets=widgets)
+    return container
+
+
+def create_form(
+    model: BaseModel | Type[BaseModel],
+    *,
+    include_fields: Optional[Iterable[str]] = None,
+    exclude_fields: Optional[Iterable[str]] = None,
+    populate_with_instance: bool = False,
+    allow_errors: bool = False,
+) -> dict[str, Any]:
+    submit_button = Button(text="Submit")
+    error_stack = Textarea(enabled=False)
+    error_sink = _ValidationErrorSink(error_stack)
+    model_widget = create_container_from_model(
+        model,
+        include_fields=include_fields,
+        exclude_fields=exclude_fields,
+        validation_error_sink=error_sink,
+        populate_with_instance=populate_with_instance,
+    )
+    container = Container(
+        widgets=[
+            submit_button,
+            model_widget,
+            error_stack,
+        ]
+    )
+    submit_button.changed.connect(container.close)
+    if isinstance(model, BaseModel):
+        container.native.setWindowTitle(model.__class__.__name__)
+    else:
+        container.native.setWindowTitle(model.__name__)
+    with event_loop():
+        container.show()
+
+    # TODO
+    # This is a big hack, but seems like investing
+    # too much time into an event driven solution is not worth it atm
+    if not allow_errors and error_sink.error is not None:
+        raise error_sink.error
+
+    return {getattr(widget[1], "name"): getattr(widget[1], "value") for widget in model_widget}