Utility Functions

NodeEncoder

Bases: JSONEncoder

Custom JSON encoder for serializing CRIPT nodes to JSON.

This encoder is used to convert CRIPT nodes into JSON format while handling unique identifiers (UUIDs) and condensed representations to avoid redundancy in the JSON output. It also allows suppressing specific attributes from being included in the serialized JSON.

Attributes:

Name	Type	Description
handled_ids	Set[str]	A set to store the UIDs of nodes that have been processed during serialization.
known_uuid	Set[str]	A set to store the UUIDs of nodes that have been previously encountered in the JSON.
condense_to_uuid	Dict[str, Set[str]]	A set to store the node types that should be condensed to UUID edges in the JSON.
suppress_attributes	Optional[Dict[str, Set[str]]]	A dictionary that allows suppressing specific attributes for nodes with the corresponding UUIDs.

Methods:

Name	Description
```python
default	Convert CRIPT nodes and other objects to their JSON representation.
```
```python
_apply_modifications	Apply modifications to the serialized dictionary based on node types and attributes to be condensed. This internal function handles node condensation and attribute suppression during serialization.
```

Source code in src/cript/nodes/util/__init__.py

class NodeEncoder(json.JSONEncoder):
    """
    Custom JSON encoder for serializing CRIPT nodes to JSON.

    This encoder is used to convert CRIPT nodes into JSON format while handling unique identifiers (UUIDs) and
    condensed representations to avoid redundancy in the JSON output.
    It also allows suppressing specific attributes from being included in the serialized JSON.

    Attributes
    ----------
    handled_ids : Set[str]
        A set to store the UIDs of nodes that have been processed during serialization.
    known_uuid : Set[str]
        A set to store the UUIDs of nodes that have been previously encountered in the JSON.
    condense_to_uuid : Dict[str, Set[str]]
        A set to store the node types that should be condensed to UUID edges in the JSON.
    suppress_attributes : Optional[Dict[str, Set[str]]]
        A dictionary that allows suppressing specific attributes for nodes with the corresponding UUIDs.

    Methods
    -------
    ```python
    default(self, obj: Any) -> Any:
        # Convert CRIPT nodes and other objects to their JSON representation.
    ```

    ```python
    _apply_modifications(self, serialize_dict: dict) -> Tuple[dict, List[str]]:
        # Apply modifications to the serialized dictionary based on node types
        # and attributes to be condensed. This internal function handles node
        # condensation and attribute suppression during serialization.
    ```
    """

    handled_ids: Set[str] = set()
    known_uuid: Set[str] = set()
    condense_to_uuid: Dict[str, Set[str]] = dict()
    suppress_attributes: Optional[Dict[str, Set[str]]] = None

    def default(self, obj):
        """
        Convert CRIPT nodes and other objects to their JSON representation.

        This method is called during JSON serialization.
        It customizes the serialization process for CRIPT nodes and handles unique identifiers (UUIDs)
        to avoid redundant data in the JSON output.
        It also allows for attribute suppression for specific nodes.

        Parameters
        ----------
        obj : Any
            The object to be serialized to JSON.

        Returns
        -------
        dict
            The JSON representation of the input object, which can be a string, a dictionary, or any other JSON-serializable type.

        Raises
        ------
        CRIPTJsonDeserializationError
            If there is an issue with the JSON deserialization process for CRIPT nodes.

        Notes
        -----
        * If the input object is a UUID, it is converted to a string representation and returned.
        * If the input object is a CRIPT node (an instance of `BaseNode`), it is serialized into a dictionary
          representation. The node is first checked for uniqueness based on its UID (unique identifier), and if
          it has already been serialized, it is represented as a UUID edge only. If not, the node's attributes
          are added to the dictionary representation, and any default attribute values are removed to reduce
          redundancy in the JSON output.
        * The method `_apply_modifications()` is called to check if further modifications are needed before
          considering the dictionary representation done. This includes condensing certain node types to UUID edges
          and suppressing specific attributes for nodes.
        """
        if isinstance(obj, uuid.UUID):
            return str(obj)
        if isinstance(obj, BaseNode):
            try:
                uid = obj.uid
            except AttributeError:
                pass
            else:
                if uid in NodeEncoder.handled_ids:
                    return {"uid": uid}

            # When saving graphs, some nodes can be pre-saved.
            # If that happens, we want to represent them as a UUID edge only
            try:
                uuid_str = str(obj.uuid)
            except AttributeError:
                pass
            else:
                if uuid_str in NodeEncoder.known_uuid:
                    return {"uuid": uuid_str}

            default_dataclass = obj.JsonAttributes()
            serialize_dict = {}
            # Remove default values from serialization
            for field_name in [field.name for field in dataclasses.fields(default_dataclass)]:
                if getattr(default_dataclass, field_name) != getattr(obj._json_attrs, field_name):
                    serialize_dict[field_name] = getattr(obj._json_attrs, field_name)
            # add the default node type
            serialize_dict["node"] = obj._json_attrs.node

            # check if further modifications to the dict is needed before considering it done
            serialize_dict, condensed_uid = self._apply_modifications(serialize_dict)
            if uid not in condensed_uid:  # We can uid (node) as handled if we don't condense it to uuid
                NodeEncoder.handled_ids.add(uid)

            # Remove suppressed attributes
            if NodeEncoder.suppress_attributes is not None and str(obj.uuid) in NodeEncoder.suppress_attributes:
                for attr in NodeEncoder.suppress_attributes[str(obj.uuid)]:
                    del serialize_dict[attr]

            return serialize_dict
        return json.JSONEncoder.default(self, obj)

    def _apply_modifications(self, serialize_dict: Dict):
        """
        Checks the serialize_dict to see if any other operations are required before it
        can be considered done. If other operations are required, then it passes it to the other operations
        and at the end returns the fully finished dict.

        This function is essentially a big switch case that checks the node type
        and determines what other operations are required for it.

        Parameters
        ----------
        serialize_dict: dict

        Returns
        -------
        serialize_dict: dict
        """

        def process_attribute(attribute):
            def strip_to_edge_uuid(element):
                # Extracts UUID and UID information from the element
                try:
                    uuid = getattr(element, "uuid")
                except AttributeError:
                    uuid = element["uuid"]
                    if len(element) == 1:  # Already a condensed element
                        return element, None
                try:
                    uid = getattr(element, "uid")
                except AttributeError:
                    uid = element["uid"]

                element = {"uuid": str(uuid)}
                return element, uid

            # Processes an attribute based on its type (list or single element)
            if isinstance(attribute, list):
                processed_elements = []
                for element in attribute:
                    processed_element, uid = strip_to_edge_uuid(element)
                    if uid is not None:
                        uid_of_condensed.append(uid)
                    processed_elements.append(processed_element)
                return processed_elements
            else:
                processed_attribute, uid = strip_to_edge_uuid(attribute)
                if uid is not None:
                    uid_of_condensed.append(uid)
                return processed_attribute

        uid_of_condensed: List = []

        nodes_to_condense = serialize_dict["node"]
        for node_type in nodes_to_condense:
            if node_type in self.condense_to_uuid:
                attributes_to_process = self.condense_to_uuid[node_type]  # type: ignore
                for attribute in attributes_to_process:
                    if attribute in serialize_dict:
                        attribute_to_condense = serialize_dict[attribute]
                        processed_attribute = process_attribute(attribute_to_condense)
                        serialize_dict[attribute] = processed_attribute

        # Check if the node is "Material" and convert the identifiers list to JSON fields
        if serialize_dict["node"] == ["Material"]:
            serialize_dict = _material_identifiers_list_to_json_fields(serialize_dict)

        return serialize_dict, uid_of_condensed

default(obj)

Convert CRIPT nodes and other objects to their JSON representation.

This method is called during JSON serialization. It customizes the serialization process for CRIPT nodes and handles unique identifiers (UUIDs) to avoid redundant data in the JSON output. It also allows for attribute suppression for specific nodes.

Parameters:

Name	Type	Description	Default
obj	Any	The object to be serialized to JSON.	required

Returns:

Type	Description
dict	The JSON representation of the input object, which can be a string, a dictionary, or any other JSON-serializable type.

Raises:

Type	Description
CRIPTJsonDeserializationError	If there is an issue with the JSON deserialization process for CRIPT nodes.

Notes

• If the input object is a UUID, it is converted to a string representation and returned.
• If the input object is a CRIPT node (an instance of BaseNode), it is serialized into a dictionary representation. The node is first checked for uniqueness based on its UID (unique identifier), and if it has already been serialized, it is represented as a UUID edge only. If not, the node's attributes are added to the dictionary representation, and any default attribute values are removed to reduce redundancy in the JSON output.
• The method _apply_modifications() is called to check if further modifications are needed before considering the dictionary representation done. This includes condensing certain node types to UUID edges and suppressing specific attributes for nodes.

Source code in src/cript/nodes/util/__init__.py

def default(self, obj):
    """
    Convert CRIPT nodes and other objects to their JSON representation.

    This method is called during JSON serialization.
    It customizes the serialization process for CRIPT nodes and handles unique identifiers (UUIDs)
    to avoid redundant data in the JSON output.
    It also allows for attribute suppression for specific nodes.

    Parameters
    ----------
    obj : Any
        The object to be serialized to JSON.

    Returns
    -------
    dict
        The JSON representation of the input object, which can be a string, a dictionary, or any other JSON-serializable type.

    Raises
    ------
    CRIPTJsonDeserializationError
        If there is an issue with the JSON deserialization process for CRIPT nodes.

    Notes
    -----
    * If the input object is a UUID, it is converted to a string representation and returned.
    * If the input object is a CRIPT node (an instance of `BaseNode`), it is serialized into a dictionary
      representation. The node is first checked for uniqueness based on its UID (unique identifier), and if
      it has already been serialized, it is represented as a UUID edge only. If not, the node's attributes
      are added to the dictionary representation, and any default attribute values are removed to reduce
      redundancy in the JSON output.
    * The method `_apply_modifications()` is called to check if further modifications are needed before
      considering the dictionary representation done. This includes condensing certain node types to UUID edges
      and suppressing specific attributes for nodes.
    """
    if isinstance(obj, uuid.UUID):
        return str(obj)
    if isinstance(obj, BaseNode):
        try:
            uid = obj.uid
        except AttributeError:
            pass
        else:
            if uid in NodeEncoder.handled_ids:
                return {"uid": uid}

        # When saving graphs, some nodes can be pre-saved.
        # If that happens, we want to represent them as a UUID edge only
        try:
            uuid_str = str(obj.uuid)
        except AttributeError:
            pass
        else:
            if uuid_str in NodeEncoder.known_uuid:
                return {"uuid": uuid_str}

        default_dataclass = obj.JsonAttributes()
        serialize_dict = {}
        # Remove default values from serialization
        for field_name in [field.name for field in dataclasses.fields(default_dataclass)]:
            if getattr(default_dataclass, field_name) != getattr(obj._json_attrs, field_name):
                serialize_dict[field_name] = getattr(obj._json_attrs, field_name)
        # add the default node type
        serialize_dict["node"] = obj._json_attrs.node

        # check if further modifications to the dict is needed before considering it done
        serialize_dict, condensed_uid = self._apply_modifications(serialize_dict)
        if uid not in condensed_uid:  # We can uid (node) as handled if we don't condense it to uuid
            NodeEncoder.handled_ids.add(uid)

        # Remove suppressed attributes
        if NodeEncoder.suppress_attributes is not None and str(obj.uuid) in NodeEncoder.suppress_attributes:
            for attr in NodeEncoder.suppress_attributes[str(obj.uuid)]:
                del serialize_dict[attr]

        return serialize_dict
    return json.JSONEncoder.default(self, obj)

add_orphaned_nodes_to_project(project, active_experiment, max_iteration=-1)

Helper function that adds all orphaned material nodes of the project graph to the project.materials attribute. Material additions only is permissible with active_experiment is None. This function also adds all orphaned data, process, computation and computational process nodes of the project graph to the active_experiment. This functions call project.validate and might raise Exceptions from there.

Source code in src/cript/nodes/util/__init__.py

def add_orphaned_nodes_to_project(project: Project, active_experiment: Experiment, max_iteration: int = -1):
    """
    Helper function that adds all orphaned material nodes of the project graph to the
    `project.materials` attribute.
    Material additions only is permissible with `active_experiment is None`.
    This function also adds all orphaned data, process, computation and computational process nodes
    of the project graph to the `active_experiment`.
    This functions call `project.validate` and might raise Exceptions from there.
    """
    if active_experiment is not None and active_experiment not in project.find_children({"node": ["Experiment"]}):
        raise RuntimeError(f"The provided active experiment {active_experiment} is not part of the project graph. Choose an active experiment that is part of a collection of this project.")

    counter = 0
    while True:
        if counter > max_iteration >= 0:
            break  # Emergency stop
        try:
            project.validate()
        except CRIPTOrphanedMaterialError as exc:
            # because calling the setter calls `validate` we have to force add the material.
            project._json_attrs.material.append(exc.orphaned_node)
        except CRIPTOrphanedDataError as exc:
            active_experiment.data += [exc.orphaned_node]
        except CRIPTOrphanedProcessError as exc:
            active_experiment.process += [exc.orphaned_node]
        except CRIPTOrphanedComputationError as exc:
            active_experiment.computation += [exc.orphaned_node]
        except CRIPTOrphanedComputationalProcessError as exc:
            active_experiment.computation_process += [exc.orphaned_node]
        else:
            break
        counter += 1

load_nodes_from_json(nodes_json)

User facing function, that return a node and all its children from a json string input.

Parameters:

Name	Type	Description	Default
nodes_json	Union[str, Dict]	JSON string representation of a CRIPT node	required

Examples:

>>> import cript
>>> # Get updated project from API
>>> my_paginator = api.search(
...     node_type=cript.Project,
...     search_mode=cript.SearchModes.EXACT_NAME,
...     value_to_search="my project name",
... )
>>> # Take specific Project you want from paginator
>>> my_project_from_api_dict: dict = my_paginator.current_page_results[0]
>>> # Deserialize your Project dict into a Project node
>>> my_project_node_from_api = cript.load_nodes_from_json(
...     nodes_json=my_project_from_api_dict
... )

Raises:

Type	Description
CRIPTJsonNodeError	If there is an issue with the JSON of the node field.
CRIPTJsonDeserializationError	If there is an error during deserialization of a specific node.
CRIPTDeserializationUIDError	If there is an issue with the UID used for deserialization, like circular references.

Notes

This function uses a custom _NodeDecoderHook to convert JSON nodes into Python objects. The _NodeDecoderHook class is responsible for handling the deserialization of nodes and caching objects with shared UIDs to avoid redundant deserialization.

The function is intended for deserializing CRIPT nodes and should not be used for generic JSON.

Returns:

Type	Description
Union[CRIPT Node, List[CRIPT Node]]	Typically returns a single CRIPT node, but if given a list of nodes, then it will serialize them and return a list of CRIPT nodes

Source code in src/cript/nodes/util/__init__.py

def load_nodes_from_json(nodes_json: Union[str, Dict]):
    """
    User facing function, that return a node and all its children from a json string input.

    Parameters
    ----------
    nodes_json: Union[str, dict]
        JSON string representation of a CRIPT node

    Examples
    --------
    >>> import cript
    >>> # Get updated project from API
    >>> my_paginator = api.search(
    ...     node_type=cript.Project,
    ...     search_mode=cript.SearchModes.EXACT_NAME,
    ...     value_to_search="my project name",
    ... ) # doctest: +SKIP
    >>> # Take specific Project you want from paginator
    >>> my_project_from_api_dict: dict = my_paginator.current_page_results[0] # doctest: +SKIP
    >>> # Deserialize your Project dict into a Project node
    >>> my_project_node_from_api = cript.load_nodes_from_json( # doctest: +SKIP
    ...     nodes_json=my_project_from_api_dict
    ... )

    Raises
    ------
    CRIPTJsonNodeError
        If there is an issue with the JSON of the node field.
    CRIPTJsonDeserializationError
        If there is an error during deserialization of a specific node.
    CRIPTDeserializationUIDError
        If there is an issue with the UID used for deserialization, like circular references.

    Notes
    -----
    This function uses a custom `_NodeDecoderHook` to convert JSON nodes into Python objects.
    The `_NodeDecoderHook` class is responsible for handling the deserialization of nodes
    and caching objects with shared UIDs to avoid redundant deserialization.

    The function is intended for deserializing CRIPT nodes and should not be used for generic JSON.

    Returns
    -------
    Union[CRIPT Node, List[CRIPT Node]]
        Typically returns a single CRIPT node,
        but if given a list of nodes, then it will serialize them and return a list of CRIPT nodes
    """
    # Initialize the custom decoder hook for JSON deserialization
    node_json_hook = _NodeDecoderHook()

    # Check if the input is already a Python dictionary
    if isinstance(nodes_json, Dict):
        # If it's a dictionary, directly use the decoder hook to deserialize it
        return node_json_hook(nodes_json)

    # Check if the input is a JSON-formatted string
    elif isinstance(nodes_json, str):
        # If it's a JSON string, parse and deserialize it using the decoder hook
        return json.loads(nodes_json, object_hook=node_json_hook)

    # Raise an error if the input type is unsupported
    else:
        raise TypeError(f"Unsupported type for nodes_json: {type(nodes_json)}")