Refactor/string paths storage rtdb

CHANGELOG.md

@@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Removed
 
 * Removed Support for IronPython 2.7 support
+* Refactored RealtimeDatabase and Storage Classes to path based references.
 * Removes dependency on `compas_fab`
 * Removed Rhino 7 post installation hooks
 

src/compas_xr/_path.py

@@ -0,0 +1,95 @@
+from typing import Union
+
+
+def normalize_path(path: Union[str, list[str], tuple[str, ...]]) -> str:
+    """Normalize a slash-delimited cloud path.
+
+    Parameters
+    ----------
+    path
+        Path as a slash-delimited string or as path segments.
+
+    Returns
+    -------
+    str
+        Normalized path with single slashes and no leading/trailing slash.
+    """
+    if isinstance(path, str):
+        raw_parts = path.strip("/").split("/")
+    elif isinstance(path, (list, tuple)):
+        raw_parts = path
+    else:
+        raise TypeError("path must be a string, list, or tuple")
+
+    parts = []
+    for part in raw_parts:
+        if not isinstance(part, str):
+            raise TypeError("all path parts must be strings")
+        stripped = part.strip("/")
+        if stripped:
+            parts.append(stripped)
+
+    return "/".join(parts)
+
+
+def path_to_parts(path: Union[str, list[str], tuple[str, ...]]) -> list[str]:
+    """Convert a path string or path parts to normalized path segments."""
+    normalized = normalize_path(path)
+    if not normalized:
+        return []
+    return normalized.split("/")
+
+
+def validate_reference_parts(parts: Union[list[str], tuple[str, ...]], invalid_chars: Union[set[str], None] = None) -> None:
+    """Validate normalized path segments for cloud references.
+
+    Parameters
+    ----------
+    parts
+        Normalized path segments.
+    invalid_chars
+        Characters that are not allowed in each path segment.
+
+    Returns
+    -------
+    None
+
+    Raises
+    ------
+    ValueError
+        If the path is empty or contains invalid characters.
+    """
+    if not parts:
+        raise ValueError("path must not be empty")
+
+    invalid_chars = invalid_chars or set()
+    for part in parts:
+        if any(char in part for char in invalid_chars):
+            raise ValueError("invalid path segment '{}': contains one of {}".format(part, " ".join(sorted(invalid_chars))))
+        if any(ord(char) < 32 or ord(char) == 127 for char in part):
+            raise ValueError("invalid path segment '{}': contains control characters".format(part))
+
+
+def validate_reference_path(path: Union[str, list[str], tuple[str, ...]], invalid_chars: Union[set[str], None] = None) -> list[str]:
+    """Normalize and validate a cloud reference path.
+
+    Parameters
+    ----------
+    path
+        Path as a slash-delimited string or as path segments.
+    invalid_chars
+        Characters that are not allowed in each path segment.
+
+    Returns
+    -------
+    list[str]
+        Normalized and validated path segments.
+
+    Raises
+    ------
+    ValueError
+        If the path is empty or contains invalid characters.
+    """
+    parts = path_to_parts(path)
+    validate_reference_parts(parts, invalid_chars=invalid_chars)
+    return parts

src/compas_xr/project/project_manager.py

@@ -104,7 +104,8 @@ def upload_data_to_project(self, data: Any, project_name: str, data_name: str) -
         data_name
             The name of the child in which data will be stored.
         """
-        self.database.upload_data_to_reference_as_child(data, project_name, data_name)
+        path = "{}/{}".format(project_name, data_name)
+        self.database.upload_data(data, path)
 
     def upload_project_data_from_compas(
         self,
@@ -143,7 +144,8 @@ def upload_qr_frames_to_project(self, project_name: str, qr_frames_list: list[Fr
         """
         qr_assembly = AssemblyExtensions().create_qr_assembly(qr_frames_list)
         data = qr_assembly.__data__
-        self.database.upload_data_to_reference_as_child(data, project_name, "QRFrames")
+        path = "{}/{}".format(project_name, "QRFrames")
+        self.database.upload_data(data, path)
 
     def upload_obj_to_storage(self, path_local: str, storage_folder_name: str) -> None:
         """
@@ -156,8 +158,9 @@ def upload_obj_to_storage(self, path_local: str, storage_folder_name: str) -> No
         storage_folder_name
             The name of the storage folder where the .obj file will be uploaded.
         """
-        storage_folder_list = ["obj_storage", storage_folder_name]
-        self.storage.upload_file_as_bytes_to_deep_reference(path_local, storage_folder_list)
+        file_name = os.path.basename(path_local)
+        storage_path = "obj_storage/{}/{}".format(storage_folder_name, file_name)
+        self.storage.upload_file_as_bytes_to_path(path_local, storage_path)
 
     def upload_objs_from_directory_to_storage(self, local_directory: str, storage_folder_name: str) -> None:
         """
@@ -170,8 +173,14 @@ def upload_objs_from_directory_to_storage(self, local_directory: str, storage_fo
         storage_folder_name
             The name of the storage folder where the .obj files will be uploaded.
         """
-        storage_folder_list = ["obj_storage", storage_folder_name]
-        self.storage.upload_files_as_bytes_from_directory_to_deep_reference(local_directory, storage_folder_list)
+        if not os.path.exists(local_directory) or not os.path.isdir(local_directory):
+            raise FileNotFoundError("Directory not found: {}".format(local_directory))
+
+        for file_name in os.listdir(local_directory):
+            local_path = os.path.join(local_directory, file_name)
+            if os.path.isfile(local_path):
+                storage_path = "obj_storage/{}/{}".format(storage_folder_name, file_name)
+                self.storage.upload_file_as_bytes_to_path(local_path, storage_path)
 
     def get_project_data(self, project_name: str) -> dict:
         """
@@ -248,13 +257,13 @@ def edit_step_on_database(
             The priority of the step.
 
         """
-        database_reference_list = [project_name, "building_plan", "data", "steps", key, "data"]
-        current_data = self.database.get_data_from_deep_reference(database_reference_list)
+        database_path = "{}/building_plan/data/steps/{}/data".format(project_name, key)
+        current_data = self.database.get_data(database_path)
         current_data["actor"] = actor
         current_data["is_built"] = is_built
         current_data["is_planned"] = is_planned
         current_data["priority"] = priority
-        self.database.upload_data_to_deep_reference(current_data, database_reference_list)
+        self.database.upload_data(current_data, database_path)
 
     def visualize_project_state_timbers(
         self,
@@ -288,8 +297,8 @@ def visualize_project_state_timbers(
 
         """
         nodes = timber_assembly.graph.__data__["node"]
-        buiding_plan_data_reference_list = [project_name, "building_plan", "data"]
-        current_state_data = self.database.get_data_from_deep_reference(buiding_plan_data_reference_list)
+        building_plan_data_path = "{}/building_plan/data".format(project_name)
+        current_state_data = self.database.get_data(building_plan_data_path)
 
         built_human = []
         unbuilt_human = []
@@ -364,8 +373,8 @@ def visualize_project_state(self, assembly: Assembly, project_name: str):
             The parts that have not been built by a robot.
 
         """
-        buiding_plan_data_reference_list = [project_name, "building_plan", "data"]
-        current_state_data = self.database.get_data_from_deep_reference(buiding_plan_data_reference_list)
+        building_plan_data_path = "{}/building_plan/data".format(project_name)
+        current_state_data = self.database.get_data(building_plan_data_path)
         nodes = assembly.graph.__data__["node"]
 
         built_human = []