GEOPY-2744: Migrate core utilities to geoapps-utils and geoh5py

geoh5py/ui_json/ui_json.py

@@ -36,9 +36,9 @@
 )
 
 from geoh5py import Workspace
-from geoh5py.groups import PropertyGroup
+from geoh5py.groups import PropertyGroup, UIJsonGroup
 from geoh5py.shared import Entity
-from geoh5py.shared.utils import fetch_active_workspace
+from geoh5py.shared.utils import fetch_active_workspace, str2uuid, stringify
 from geoh5py.shared.validators import none_to_empty_string
 from geoh5py.ui_json.forms import BaseForm
 from geoh5py.ui_json.validations import ErrorPool, UIJsonError, get_validations
@@ -67,7 +67,9 @@ class BaseUIJson(BaseModel):
     :params workspace_geoh5: Path to the workspace geoh5 file.
     """
 
-    model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True, extra="allow", validate_assignment=True
+    )
 
     version: str
     title: str
@@ -90,7 +92,7 @@ def __str__(self) -> str:
         """String level shows the full json representation."""
 
         json_string = self.model_dump_json(indent=4, exclude_unset=True)
-        for field in self.model_fields:
+        for field in type(self).model_fields:
             value = getattr(self, field)
             if isinstance(value, BaseForm):
                 type_string = type(value).__name__
@@ -102,15 +104,15 @@ def __str__(self) -> str:
 
     @field_validator("geoh5", mode="after")
     @classmethod
-    def workspace_path_exists(cls, path: Path):
-        if not path.exists():
+    def workspace_path_exists(cls, path: Path | None) -> Path | None:
+        if path is not None and not path.exists():
             raise FileNotFoundError(f"geoh5 path {path} does not exist.")
         return path
 
     @field_validator("geoh5", mode="after")
     @classmethod
-    def valid_geoh5_extension(cls, path: Path):
-        if path.suffix != ".geoh5":
+    def valid_geoh5_extension(cls, path: Path | None) -> Path | None:
+        if path is not None and path.suffix != ".geoh5":
             raise ValueError(
                 f"Workspace path: {path} must have a '.geoh5' file extension."
             )
@@ -143,6 +145,9 @@ def read(cls, path: str | Path) -> BaseUIJson:
 
         with open(path, encoding="utf-8") as file:
             kwargs = json.load(file)
+            kwargs = {
+                key: (item if item != "" else None) for key, item in kwargs.items()
+            }
 
         if cls == BaseUIJson:
             fields = {}
@@ -250,6 +255,49 @@ def flatten(self, skip_disabled=False, active_only=False) -> dict[str, Any]:
 
         return data
 
+    def fill(self, copy: bool = False, **kwargs) -> BaseUIJson:
+        """
+        Fill the UIJson with new values.
+
+        :param copy: If True, returns a new UIJson object with the updated values.
+            If False, updates the current UIJson object with the new values and returns itself.
+        :param kwargs: Key/value pairs to update the UIJson with.
+
+        :return: A new UIJson object with the updated values.
+        """
+        temp_properties = {}
+        for key, form in dict(self).items():
+            if not isinstance(form, BaseForm):
+                if key in kwargs:
+                    if not isinstance(kwargs[key], str):
+                        raise TypeError(
+                            "Only string values can be updated for non-form fields. "
+                        )
+                    temp_properties[key] = kwargs[key]
+                continue
+
+            updates: dict[str, Any] = {}
+
+            # if a value has no default value, set enabled to false
+            if not bool(form.value) if form.value != [""] else False:
+                updates["enabled"] = False
+
+            if key in kwargs:
+                updates["value"] = str2uuid(stringify(kwargs[key]))
+                updates["enabled"] = True
+
+            if updates:
+                temp_properties[key] = form.model_copy(update=updates)
+
+        updated_model = self.model_copy(update=temp_properties)
+
+        if not copy:
+            for field_name in type(self).model_fields:
+                setattr(self, field_name, getattr(updated_model, field_name))
+            return self
+
+        return updated_model
+
     def to_params(self, workspace: Workspace | None = None) -> dict[str, Any]:
         """
         Promote, flatten and validate parameter/values dictionary.
@@ -287,6 +335,35 @@ def to_params(self, workspace: Workspace | None = None) -> dict[str, Any]:
 
         return data
 
+    def to_ui_json_group(
+        self, workspace: Workspace | None = None, **kwargs
+    ) -> UIJsonGroup:
+        """
+        Convert the UIJson to a UIJsonGroup.
+
+        :param workspace: Workspace to fetch entities from.  Used for passing active
+            workspaces to avoid closing and flushing data.
+        :param kwargs: Additional keyword arguments to update the UIJson data before
+
+        :return: A UIJsonGroup representing the application.
+        """
+        with fetch_active_workspace(workspace or Workspace(self.geoh5)) as geoh5:
+            if geoh5 is None:
+                raise ValueError("Workspace cannot be None.")
+
+            ui_json_group = UIJsonGroup.create(
+                workspace=geoh5,
+                options=self.model_dump(mode="json", exclude_unset=True, by_alias=True),
+                name=kwargs.pop("name", self.title),
+                **kwargs,
+            )
+            options = ui_json_group.options
+            options["out_group"]["value"] = ui_json_group.uid
+            options["out_group"]["enabled"] = True
+            ui_json_group.options = options
+
+            return ui_json_group
+
     def validate_data(
         self, params: dict[str, Any] | None = None, errors: dict[str, Any] | None = None
     ) -> None:

tests/ui_json/uijson_test.py

@@ -28,6 +28,7 @@
 from pydantic import ValidationError
 
 from geoh5py import Workspace
+from geoh5py.groups import UIJsonGroup
 from geoh5py.objects import Curve, Points
 from geoh5py.ui_json.annotations import Deprecated
 from geoh5py.ui_json.forms import (
@@ -589,3 +590,221 @@ def test_geoh5_validate_extension(tmp_path):
             conda_environment="test",
             workspace_geoh5=None,
         )
+
+
+# ---------------------------------------------------------------------------
+# fill tests
+# ---------------------------------------------------------------------------
+
+
+def test_fill_in_place(tmp_path):
+    ws = Workspace(tmp_path / "test.geoh5")
+
+    class MyUIJson(BaseUIJson):
+        my_string_parameter: StringForm
+        my_int_parameter: IntegerForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={
+            "my_string_parameter": {"label": "a", "value": "original"},
+            "my_int_parameter": {"label": "b", "value": 1},
+        },
+    )
+    result = uijson.fill(my_string_parameter="updated")
+
+    assert result is uijson
+    assert uijson.my_string_parameter.value == "updated"
+    assert uijson.my_int_parameter.value == 1
+
+
+def test_fill_copy(tmp_path):
+    ws = Workspace(tmp_path / "test.geoh5")
+
+    class MyUIJson(BaseUIJson):
+        my_string_parameter: StringForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={"my_string_parameter": {"label": "a", "value": "original"}},
+    )
+    copy = uijson.fill(copy=True, my_string_parameter="updated", title="ok")
+
+    assert copy is not uijson
+    assert copy.my_string_parameter.value == "updated"
+    assert uijson.my_string_parameter.value == "original"
+    assert copy.title == "ok"
+
+    with pytest.raises(TypeError, match="Only string"):
+        _ = uijson.fill(copy=True, my_string_parameter="updated", title=666)
+
+
+def test_fill_disables_forms_with_falsy_value(tmp_path):
+    ws = Workspace(tmp_path / "test.geoh5")
+
+    class MyUIJson(BaseUIJson):
+        my_zero_param: FloatForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={"my_zero_param": {"label": "a", "value": 0.0}},
+    )
+    uijson.fill()
+
+    assert uijson.my_zero_param.enabled is False
+
+
+def test_fill_truthy_value_leaves_updates_empty(tmp_path):
+    """A form with a truthy value not in kwargs produces no updates."""
+    ws = Workspace(tmp_path / "test.geoh5")
+
+    class MyUIJson(BaseUIJson):
+        my_param: FloatForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={"my_param": {"label": "a", "value": 3.14}},
+    )
+    original_enabled = uijson.my_param.enabled
+    uijson.fill()
+
+    assert uijson.my_param.enabled == original_enabled
+    assert uijson.my_param.value == 3.14
+
+
+def test_fill_kwargs_re_enables_form(tmp_path):
+    ws = Workspace(tmp_path / "test.geoh5")
+
+    class MyUIJson(BaseUIJson):
+        my_param: FloatForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={"my_param": {"label": "a", "value": 0.0, "enabled": False}},
+    )
+    uijson.fill(my_param=5.0)
+
+    assert uijson.my_param.enabled is True
+    assert uijson.my_param.value == 5.0
+
+
+def test_fill_with_uuid_value(tmp_path):
+    ws = Workspace(tmp_path / "test.geoh5")
+    pts = Points.create(ws, name="pts", vertices=np.random.random((10, 3)))
+    pts2 = Points.create(ws, name="pts2", vertices=np.random.random((10, 3)))
+
+    class MyUIJson(BaseUIJson):
+        my_object_parameter: ObjectForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={
+            "my_object_parameter": {
+                "label": "obj",
+                "mesh_type": [Points],
+                "value": pts.uid,
+            }
+        },
+    )
+    uijson.fill(my_object_parameter=pts2.uid)
+
+    assert uijson.my_object_parameter.value == pts2.uid
+
+
+# ---------------------------------------------------------------------------
+# to_ui_json_group tests
+# ---------------------------------------------------------------------------
+
+
+def test_to_ui_json_group_creates_group(tmp_path):
+    ws = Workspace.create(tmp_path / "test.geoh5")
+
+    class MyUIJson(BaseUIJson):
+        my_string_parameter: StringForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={"my_string_parameter": {"label": "a", "value": "test"}},
+    )
+    group = uijson.to_ui_json_group(workspace=ws)
+
+    assert isinstance(group, UIJsonGroup)
+    assert ws.get_entity(group.uid)[0] is not None
+
+
+def test_to_ui_json_group_default_name(tmp_path):
+    ws = Workspace.create(tmp_path / "test.geoh5")
+
+    class MyUIJson(BaseUIJson):
+        my_string_parameter: StringForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={"my_string_parameter": {"label": "a", "value": "test"}},
+    )
+    group = uijson.to_ui_json_group(workspace=ws)
+
+    assert group.name == "my application"
+
+
+def test_to_ui_json_group_custom_name(tmp_path):
+    ws = Workspace.create(tmp_path / "test.geoh5")
+
+    class MyUIJson(BaseUIJson):
+        my_string_parameter: StringForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={"my_string_parameter": {"label": "a", "value": "test"}},
+    )
+    group = uijson.to_ui_json_group(workspace=ws, name="custom name")
+
+    assert group.name == "custom name"
+
+
+def test_to_ui_json_group_out_group_properties(tmp_path):
+    ws = Workspace.create(tmp_path / "test.geoh5")
+
+    class MyUIJson(BaseUIJson):
+        my_string_parameter: StringForm
+
+    uijson = generate_test_uijson(
+        ws,
+        uijson=MyUIJson,
+        data={"my_string_parameter": {"label": "a", "value": "test"}},
+    )
+    group = uijson.to_ui_json_group(workspace=ws)
+
+    assert group.options["out_group"]["value"] == str(group.uid)
+    assert group.options["out_group"]["enabled"] is True
+
+
+def test_to_ui_json_group_without_workspace(tmp_path):
+    geoh5_path = tmp_path / "test.geoh5"
+    Workspace.create(geoh5_path)
+
+    class MyUIJson(BaseUIJson):
+        my_string_parameter: StringForm
+
+    uijson = MyUIJson(
+        version="0.1.0",
+        title="my application",
+        geoh5=str(geoh5_path),
+        run_command="python -m my_module",
+        monitoring_directory=None,
+        conda_environment="test",
+        workspace_geoh5=None,
+        my_string_parameter={"label": "a", "value": "test"},
+    )
+    group = uijson.to_ui_json_group()
+
+    assert isinstance(group, UIJsonGroup)