capturegraph.procedures.procedure

Core Procedure Framework

This module defines the base Procedure class and supporting infrastructure for the CaptureGraph library. The Procedure class is the foundation for all workflow nodes, providing type safety, validation, serialization, and fluent API capabilities.

Key Features:

• Generic typing system with return type validation

• Automatic input and substep type checking

• Unique identification and graph serialization

• Fluent API for chaining procedure operations

• Extension methods for common operations

All procedure nodes in CaptureGraph extend this base class to inherit these capabilities.

Procedure

Base class for all procedures in the CaptureGraph library.

Procedures represent nodes in a directed acyclic graph (DAG) that define actions and transformations to be executed by the app. Each procedure has a return type T that extends PType, indicating what kind of data it produces.

Key features: - Type safety through generic return types - Automatic validation of inputs and substeps - Unique identification via UUIDs - Serialization to JSON for app execution - Fluent API for building procedure chains

Attributes:

Name	Type	Description
label	Optional[str]	Optional human-readable description of what this procedure does
_kind	type	The specific procedure class type
_return_type	type[PType]	The PType that this procedure returns
_uuid	str	Unique identifier for this procedure instance

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

@make_procedure
class Procedure[T: PType]:
    """
    Base class for all procedures in the CaptureGraph library.

    Procedures represent nodes in a directed acyclic graph (DAG) that define
    actions and transformations to be executed by the app. Each procedure
    has a return type T that extends PType, indicating what kind of data it produces.

    Key features:
    - Type safety through generic return types
    - Automatic validation of inputs and substeps
    - Unique identification via UUIDs
    - Serialization to JSON for app execution
    - Fluent API for building procedure chains

    Attributes:
        label: Optional human-readable description of what this procedure does
        _kind: The specific procedure class type
        _return_type: The PType that this procedure returns
        _uuid: Unique identifier for this procedure instance
    """

    label: Optional[str] = field(default=None, kw_only=True)

    _kind: type = field(init=False, repr=False)
    _return_type: type[PType] = field(init=False, repr=False)
    _uuid: str = field(default_factory=lambda: str(uuid4()), init=False, repr=False)

    ### Initialization and subclassing ###

    def __post_init__(self) -> None:
        """Validate all procedure inputs, substeps, and settings at initialization."""
        self._validate_input_procedures()
        self._validate_substep_procedures()
        self._validate_setting_types()

    def _validate_input_procedures(self) -> None:
        """Ensure all input procedures return types compatible with field annotations."""
        for field_name, input_procedure in self._inputs.items():
            # Use the procedure's resolved _return_type which is more accurate than parsing generics
            actual_return_type = input_procedure._return_type

            field_type = self.__dataclass_fields__[field_name].type
            expected_return_type = get_args(field_type)[0]

            if not self._is_compatible_procedure(
                actual_return_type, expected_return_type
            ):
                raise TypeError(
                    f"Input '{field_name}' for {self._kind.__name__} expects {expected_return_type.__name__}, "
                    f"but got {actual_return_type.__name__} from {input_procedure}"
                )

    def _validate_substep_procedures(self) -> None:
        """Ensure all substeps return PVoid (side effects only, no data flow)."""
        for substep in self._substeps:
            # Use the substep's resolved _return_type directly - most accurate approach
            substep_return_type = substep._return_type
            if substep_return_type is not PVoid:
                raise TypeError(
                    f"Substep {substep} in {self._kind.__name__} must return PVoid, "
                    f"but returns {substep_return_type.__name__}"
                )

    def _validate_setting_types(self) -> None:
        """Validate that all non-procedure settings match their field type annotations."""
        for field_name, setting_value in self._settings.items():
            expected_type = self.__dataclass_fields__[field_name].type
            if not self._is_valid_setting_type(setting_value, expected_type):
                raise TypeError(
                    f"Setting '{field_name}' for {self._kind.__name__} expects {expected_type}, "
                    f"but got {type(setting_value)} with value: {setting_value}"
                )

    def _is_compatible_procedure(self, actual_type: type, expected_type: type) -> bool:
        """
        Check if actual_type is compatible with expected_type using inheritance.

        This method prioritizes the resolved _return_type from Procedure instances
        over parsing generic type annotations, following the same philosophy as
        the forward_types decorator.
        """
        try:
            if not isinstance(expected_type, type):
                return issubclass(actual_type, self._return_type)

            return issubclass(actual_type, expected_type)
        except:
            return False

    def _is_valid_setting_type(self, value: Any, expected_type: type) -> bool:
        """Validate a setting value against its expected type, handling lists."""
        try:
            if isinstance(value, list):
                return self._is_valid_list_setting(value, expected_type)

            return isinstance(value, expected_type)
        except:
            return False

    def _is_valid_list_setting(self, value_list: list, expected_type: type) -> bool:
        """Validate that all items in a list setting match the expected element type."""
        try:
            if not hasattr(expected_type, "__args__") or not expected_type.__args__:
                return False

            expected_element_type = get_args(expected_type)[0]
            return all(isinstance(item, expected_element_type) for item in value_list)
        except:
            return False

    def __init_subclass__(cls) -> None:
        super().__init_subclass__()

        for base in getattr(cls, "__orig_bases__", ()):
            if get_origin(base) is Procedure:
                rt = get_args(base)[0]
                cls._kind = cls
                cls._return_type = rt
                break

    ### Various useful overrides ###

    def __hash__(self) -> int:
        return hash(self._uuid)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Procedure):
            return False
        return self._uuid == other._uuid

    def __str__(self) -> str:
        if self.label:
            return f"{self.label} ({self._kind.__name__})"
        else:
            return f"{self._kind.__name__}"

    ### Properties for accessing procedure context ###

    @property
    def _context(self) -> dict[str, Any]:
        return {
            name: getattr(self, name, None)
            for name, fld in self.__dataclass_fields__.items()
            if name not in ("label") and not name.startswith("_")
        }

    @staticmethod
    def _is_procedure(value: Any) -> bool:
        return isinstance(value, Procedure) and issubclass(value._return_type, PType)

    @property
    def _inputs(self) -> dict[str, "Procedure"]:
        return {
            name: value
            for name, value in self._context.items()
            if Procedure._is_procedure(value)
        }

    @staticmethod
    def _is_procedure_list(value: Any) -> bool:
        return isinstance(value, list) and all(
            Procedure._is_procedure(v) for v in value
        )

    @property
    def _substeps(self) -> list["Procedure"]:
        return [
            v
            for value in self._context.values()
            if Procedure._is_procedure_list(value)
            for v in value
        ]

    @staticmethod
    def _is_setting(value: Any) -> bool:
        return (
            not Procedure._is_procedure(value)
            and not Procedure._is_procedure_list(value)
            and value is not None
        )

    @property
    def _settings(self) -> dict[str, Any]:
        return {
            name: value
            for name, value in self._context.items()
            if Procedure._is_setting(value)
        }

    ##### Convenience methods for changing labels #####

    def set_label(self: "Procedure[T]", label: str) -> "Procedure[T]":
        """Sets a new label for this procedure."""
        object.__setattr__(self, "label", label)
        return self

    ##### Convenience methods for creating new procedures #####

    # Requirements and Optionals

    def make_optional(self: "Procedure[PVoid]") -> "Procedure[PVoid]":
        """Shortcut for OptionalProcedure node."""
        ...

    def require(self: "Procedure[PType]") -> "Procedure[PVoid]":
        """Shortcut for RequireProcedureCompleted node."""
        ...

    # Directory and File creation methods

    def new_directory(
        self: "Procedure[PDirectory]", name: str
    ) -> "Procedure[PDirectory]":
        """
        Creates a new subdirectory within this directory.

        Creates a procedure that will create a new subdirectory for organizing
        captured data. The directory will be created if it doesn't exist, or
        reused if it already exists. Does not overwrite existing directories.

        Args:
            name: The name of the new directory to create

        Returns:
            A procedure that returns a PDirectory reference to the subdirectory

        Example:
            ```python
            session = target.new_session("daily_capture")
            data_folder = session.new_directory("measurements")
            ```
        """
        ...

    def new_session(
        self: "Procedure[PDirectory]", name: str
    ) -> "Procedure[PDirectory]":
        """
        Creates a new timestamped session directory for this execution.

        Creates a procedure that will create a new session with a unique
        hex-encoded microsecond timestamp. Sessions enable multiple executions
        of the same procedure within a target, with each execution's data
        organized separately and chronologically.

        Args:
            name: The session type name (e.g., "daily_capture", "field_measurement")

        Returns:
            A procedure that returns a PDirectory reference to the session directory

        Example:
            ```python
            target = GetRootDirectory()
            session = target.new_session("daily_measurement")
            # Creates: sessions/daily_measurement/00063B40E29D696A/
            ```
        """
        ...

    def new_file(self: "Procedure[PDirectory]", name: str) -> "Procedure[PFile]":
        """
        Creates a file reference within this directory.

        Creates a procedure that returns a reference to a file that can be used
        with save(), cache(), or load() operations. The file itself is not created
        until data is actually saved to it.

        Args:
            name: The name of the file to create a reference for

        Returns:
            A procedure that returns a PFile reference to the file

        Example:
            ```python
            session = target.new_session("daily_capture")
            photo_file = session.new_file("daily_photo")
            photo_file.save(CaptureImage(label="Today's photo"))
            ```
        """
        ...

    def get_first_session(
        self: "Procedure[PDirectory]", name: str
    ) -> "Procedure[PDirectory]":
        """Shortcut for GetFirstSessionDirectory node."""
        ...

    def get_last_session(
        self: "Procedure[PDirectory]", name: str
    ) -> "Procedure[PDirectory]":
        """Shortcut for GetLastSessionDirectory node."""
        ...

    def get_user_directory(
        self: "Procedure[PDirectory]",
        name: str,
        user: "Optional[Procedure[PUserID]]" = None,
    ) -> "Procedure[PDirectory]":
        """
        Creates a user-specific subdirectory for per-user configurations.

        Creates a procedure that creates a subdirectory with the structure:
        {parent}/{name}/{user.identifier}/. Useful for collaborative capture
        scenarios where multiple users share a target but need separate
        configuration directories.

        Args:
            name: The base directory name (e.g., "config", "preferences")
            user: Optional user ID procedure. Defaults to GetUserID() if not provided.

        Returns:
            A procedure that returns a PDirectory reference to the user's directory

        Example:
            ```python
            target = GetRootDirectory()
            user_config = target.get_user_directory("config")
            # Creates: config/A1B2C3D4-E5F6.../
            ```
        """
        ...

    # Directory Metadata Methods

    def get_thumbnail_file(self: "Procedure[PDirectory]") -> "Procedure[PFile]":
        """Shortcut for GetThumbnailFile node."""
        ...

    def get_location_file(self: "Procedure[PDirectory]") -> "Procedure[PFile]":
        """Shortcut for GetLocationFile node."""
        ...

    def get_notification_file(self: "Procedure[PDirectory]") -> "Procedure[PFile]":
        """Shortcut for GetNotificationFile node."""
        ...

    def mark_for_clear_after_backup(
        self: "Procedure[PDirectory]",
    ) -> "Procedure[PVoid]":
        """Shortcut for MarkForClearAfterBackup node."""
        ...

    # Saving and Caching methods

    def save(
        self: "Procedure[PFile]",
        procedure: "Procedure[PSaveable]",
        skip_if_exists: bool = False,
    ) -> "Procedure[PVoid]":
        """
        Saves data from a procedure to this file and returns PVoid.

        Creates a procedure that stores the output of the given procedure to this
        file. Unlike cache(), this returns PVoid, making it suitable for use in
        ProcedureSequence workflows where no return value is needed.

        The data is persisted at the end of procedure execution, after the user
        indicates they are done. If the user cancels early, no partial data is saved.

        Args:
            procedure: The procedure whose output should be saved to this file
            skip_if_exists: Whether to skip procedure if the file already exists

        Returns:
            A procedure that returns PVoid after saving is complete

        Example:
            ```python
            session = target.new_session("daily_capture")
            session.new_file("photo").save(CaptureImage(label="Daily Photo"))
            ```
        """
        ...

    def save_bundle(
        self: "Procedure[PDirectory]",
        procedure: "Procedure[PSaveableBundle]",
        skip_if_exists: bool = False,
    ) -> "Procedure[PVoid]":
        """
        Saves data from a procedure to this folder and returns PVoid.

        Creates a procedure that stores the output of the given procedure to this
        folder. Unlike cache_bundle(), this returns PVoid, making it suitable for use in
        ProcedureSequence workflows where no return value is needed.

        The data is persisted at the end of procedure execution, after the user
        indicates they are done. If the user cancels early, no partial data is saved.

        Args:
            procedure: The procedure whose output should be saved to this folder
            skip_if_exists: Whether to skip procedure if the folder already exists

        Returns:
            A procedure that returns PVoid after saving is complete

        Example:
            ```python
            session = target.new_session("daily_capture")
            session.new_folder("photos").save_bundle(CaptureImageSequence(label="Daily Photos"))
            ```
        """
        ...

    def cache[T2: PSaveable](
        self: "Procedure[PFile]",
        procedure: "Procedure[T2]",
        skip_if_exists: bool = False,
    ) -> "Procedure[T2]":
        """
        Caches data from a procedure to this file and returns the original data type.

        Creates a procedure that stores the output of the given procedure to this
        file AND returns the data for continued use in the workflow. This is useful
        for persistent configuration data that you want to use in future sessions
        and also reference in conditional logic.

        The data is persisted at the end of procedure execution. If skip_if_exists
        is True and the file exists, the existing data is preloaded for user editing.

        Args:
            procedure: The procedure whose output should be cached to this file
            skip_if_exists: Whether to skip procedure if the file already exists

        Returns:
            A procedure that returns the same type as the input procedure

        Example:
            ```python
            user_setting = target.new_file("setting").cache(
                UserInputString(label="Your preference"),
                skip_if_exists=True
            )
            # Now user_setting can be used in conditional logic
            ```
        """
        ...

    def cache_bundle[T2: PSaveableBundle](
        self: "Procedure[PDirectory]",
        procedure: "Procedure[T2]",
        skip_if_exists: bool = False,
    ) -> "Procedure[T2]":
        """
        Caches data from a procedure to this folder and returns the original data type.

        Creates a procedure that stores the output of the given procedure to this
        folder AND returns the data for continued use in the workflow. This is useful
        for persistent configuration data that you want to use in future sessions
        and also reference in conditional logic.

        The data is persisted at the end of procedure execution. If skip_if_exists
        is True and the folder exists, the existing data is preloaded for user editing.

        Args:
            procedure: The procedure whose output should be cached to this folder
            skip_if_exists: Whether to skip procedure if the folder already exists

        Returns:
            A procedure that returns the same type as the input procedure
        """
        ...

    def load[T2: PSaveable](
        self: "Procedure[PFile]",
        or_else: "Procedure[T2]",
    ) -> "Procedure[T2]":
        """
        Loads existing data from this file or executes fallback procedure.

        Creates a procedure that tries to load previously saved data from this file.
        If the file doesn't exist or has no data, it executes the fallback procedure
        instead. This is commonly used for reference-based workflows where you want
        to load existing references or create new ones.

        Args:
            or_else: The fallback procedure to execute if no data exists in the file

        Returns:
            A procedure that returns the loaded data or fallback result

        Example:
            ```python
            reference = target.new_file("reference_image").load(
                or_else=CaptureImage(label="Set baseline photo")
            )
            # Returns loaded image if exists, or prompts for new capture
            ```
        """
        ...

    def load_bundle[T2: PSaveableBundle](
        self: "Procedure[PDirectory]",
        or_else: "Procedure[T2]",
    ) -> "Procedure[T2]":
        """Shortcut for CacheProcedureToBundle node (load mode)."""
        ...

    # Boolean methods

    def __or__(
        self: "Procedure[PBool]", other: "Procedure[PBool]"
    ) -> "Procedure[PBool]":
        """Shortcut for BoolOr node."""
        ...

    def __and__(
        self: "Procedure[PBool]", other: "Procedure[PBool]"
    ) -> "Procedure[PBool]":
        """Shortcut for BoolAnd node."""
        ...

    def __invert__(self: "Procedure[PBool]") -> "Procedure[PBool]":
        """Shortcut for BoolNot node."""
        ...

    # Number methods

    def __lt__(
        self: "Procedure[PNumber]", other: "Procedure[PNumber]"
    ) -> "Procedure[PBool]":
        """Shortcut for NumberLessThan node."""
        ...

    def __gt__(
        self: "Procedure[PNumber]", other: "Procedure[PNumber]"
    ) -> "Procedure[PBool]":
        """Shortcut for NumberLessThan node (with operands swapped)."""
        return other < self

    def __le__(
        self: "Procedure[PNumber]", other: "Procedure[PNumber]"
    ) -> "Procedure[PBool]":
        """Shortcut for BoolNot(NumberLessThan) node combination."""
        return ~(other < self)

    def __ge__(
        self: "Procedure[PNumber]", other: "Procedure[PNumber]"
    ) -> "Procedure[PBool]":
        """Shortcut for BoolNot(NumberLessThan) node combination (with operands swapped)."""
        return ~(self < other)

__post_init__()

Validate all procedure inputs, substeps, and settings at initialization.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def __post_init__(self) -> None:
    """Validate all procedure inputs, substeps, and settings at initialization."""
    self._validate_input_procedures()
    self._validate_substep_procedures()
    self._validate_setting_types()

set_label(label)

Sets a new label for this procedure.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def set_label(self: "Procedure[T]", label: str) -> "Procedure[T]":
    """Sets a new label for this procedure."""
    object.__setattr__(self, "label", label)
    return self

make_optional()

Shortcut for OptionalProcedure node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def make_optional(self: "Procedure[PVoid]") -> "Procedure[PVoid]":
    """Shortcut for OptionalProcedure node."""
    ...

require()

Shortcut for RequireProcedureCompleted node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def require(self: "Procedure[PType]") -> "Procedure[PVoid]":
    """Shortcut for RequireProcedureCompleted node."""
    ...

new_directory(name)

Creates a new subdirectory within this directory.

Creates a procedure that will create a new subdirectory for organizing captured data. The directory will be created if it doesn't exist, or reused if it already exists. Does not overwrite existing directories.

Parameters:

Name	Type	Description	Default
name	str	The name of the new directory to create	required

Returns:

Type	Description
Procedure[PDirectory]	A procedure that returns a PDirectory reference to the subdirectory

Example

session = target.new_session("daily_capture")
data_folder = session.new_directory("measurements")

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def new_directory(
    self: "Procedure[PDirectory]", name: str
) -> "Procedure[PDirectory]":
    """
    Creates a new subdirectory within this directory.

    Creates a procedure that will create a new subdirectory for organizing
    captured data. The directory will be created if it doesn't exist, or
    reused if it already exists. Does not overwrite existing directories.

    Args:
        name: The name of the new directory to create

    Returns:
        A procedure that returns a PDirectory reference to the subdirectory

    Example:
        ```python
        session = target.new_session("daily_capture")
        data_folder = session.new_directory("measurements")
        ```
    """
    ...

new_session(name)

Creates a new timestamped session directory for this execution.

Creates a procedure that will create a new session with a unique hex-encoded microsecond timestamp. Sessions enable multiple executions of the same procedure within a target, with each execution's data organized separately and chronologically.

Parameters:

Name	Type	Description	Default
name	str	The session type name (e.g., "daily_capture", "field_measurement")	required

Returns:

Type	Description
Procedure[PDirectory]	A procedure that returns a PDirectory reference to the session directory

Example

target = GetRootDirectory()
session = target.new_session("daily_measurement")
# Creates: sessions/daily_measurement/00063B40E29D696A/

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def new_session(
    self: "Procedure[PDirectory]", name: str
) -> "Procedure[PDirectory]":
    """
    Creates a new timestamped session directory for this execution.

    Creates a procedure that will create a new session with a unique
    hex-encoded microsecond timestamp. Sessions enable multiple executions
    of the same procedure within a target, with each execution's data
    organized separately and chronologically.

    Args:
        name: The session type name (e.g., "daily_capture", "field_measurement")

    Returns:
        A procedure that returns a PDirectory reference to the session directory

    Example:
        ```python
        target = GetRootDirectory()
        session = target.new_session("daily_measurement")
        # Creates: sessions/daily_measurement/00063B40E29D696A/
        ```
    """
    ...

new_file(name)

Creates a file reference within this directory.

Creates a procedure that returns a reference to a file that can be used with save(), cache(), or load() operations. The file itself is not created until data is actually saved to it.

Parameters:

Name	Type	Description	Default
name	str	The name of the file to create a reference for	required

Returns:

Type	Description
Procedure[PFile]	A procedure that returns a PFile reference to the file

Example

session = target.new_session("daily_capture")
photo_file = session.new_file("daily_photo")
photo_file.save(CaptureImage(label="Today's photo"))

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def new_file(self: "Procedure[PDirectory]", name: str) -> "Procedure[PFile]":
    """
    Creates a file reference within this directory.

    Creates a procedure that returns a reference to a file that can be used
    with save(), cache(), or load() operations. The file itself is not created
    until data is actually saved to it.

    Args:
        name: The name of the file to create a reference for

    Returns:
        A procedure that returns a PFile reference to the file

    Example:
        ```python
        session = target.new_session("daily_capture")
        photo_file = session.new_file("daily_photo")
        photo_file.save(CaptureImage(label="Today's photo"))
        ```
    """
    ...

get_first_session(name)

Shortcut for GetFirstSessionDirectory node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def get_first_session(
    self: "Procedure[PDirectory]", name: str
) -> "Procedure[PDirectory]":
    """Shortcut for GetFirstSessionDirectory node."""
    ...

get_last_session(name)

Shortcut for GetLastSessionDirectory node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def get_last_session(
    self: "Procedure[PDirectory]", name: str
) -> "Procedure[PDirectory]":
    """Shortcut for GetLastSessionDirectory node."""
    ...

get_user_directory(name, user=None)

Creates a user-specific subdirectory for per-user configurations.

Creates a procedure that creates a subdirectory with the structure: {parent}/{name}/{user.identifier}/. Useful for collaborative capture scenarios where multiple users share a target but need separate configuration directories.

Parameters:

Name	Type	Description	Default
name	str	The base directory name (e.g., "config", "preferences")	required
user	Optional[Procedure[PUserID]]	Optional user ID procedure. Defaults to GetUserID() if not provided.	None

Returns:

Type	Description
Procedure[PDirectory]	A procedure that returns a PDirectory reference to the user's directory

Example

target = GetRootDirectory()
user_config = target.get_user_directory("config")
# Creates: config/A1B2C3D4-E5F6.../

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def get_user_directory(
    self: "Procedure[PDirectory]",
    name: str,
    user: "Optional[Procedure[PUserID]]" = None,
) -> "Procedure[PDirectory]":
    """
    Creates a user-specific subdirectory for per-user configurations.

    Creates a procedure that creates a subdirectory with the structure:
    {parent}/{name}/{user.identifier}/. Useful for collaborative capture
    scenarios where multiple users share a target but need separate
    configuration directories.

    Args:
        name: The base directory name (e.g., "config", "preferences")
        user: Optional user ID procedure. Defaults to GetUserID() if not provided.

    Returns:
        A procedure that returns a PDirectory reference to the user's directory

    Example:
        ```python
        target = GetRootDirectory()
        user_config = target.get_user_directory("config")
        # Creates: config/A1B2C3D4-E5F6.../
        ```
    """
    ...

get_thumbnail_file()

Shortcut for GetThumbnailFile node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def get_thumbnail_file(self: "Procedure[PDirectory]") -> "Procedure[PFile]":
    """Shortcut for GetThumbnailFile node."""
    ...

get_location_file()

Shortcut for GetLocationFile node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def get_location_file(self: "Procedure[PDirectory]") -> "Procedure[PFile]":
    """Shortcut for GetLocationFile node."""
    ...

get_notification_file()

Shortcut for GetNotificationFile node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def get_notification_file(self: "Procedure[PDirectory]") -> "Procedure[PFile]":
    """Shortcut for GetNotificationFile node."""
    ...

mark_for_clear_after_backup()

Shortcut for MarkForClearAfterBackup node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def mark_for_clear_after_backup(
    self: "Procedure[PDirectory]",
) -> "Procedure[PVoid]":
    """Shortcut for MarkForClearAfterBackup node."""
    ...

save(procedure, skip_if_exists=False)

Saves data from a procedure to this file and returns PVoid.

Creates a procedure that stores the output of the given procedure to this file. Unlike cache(), this returns PVoid, making it suitable for use in ProcedureSequence workflows where no return value is needed.

The data is persisted at the end of procedure execution, after the user indicates they are done. If the user cancels early, no partial data is saved.

Parameters:

Name	Type	Description	Default
procedure	Procedure[PSaveable]	The procedure whose output should be saved to this file	required
skip_if_exists	bool	Whether to skip procedure if the file already exists	False

Returns:

Type	Description
Procedure[PVoid]	A procedure that returns PVoid after saving is complete

Example

session = target.new_session("daily_capture")
session.new_file("photo").save(CaptureImage(label="Daily Photo"))

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def save(
    self: "Procedure[PFile]",
    procedure: "Procedure[PSaveable]",
    skip_if_exists: bool = False,
) -> "Procedure[PVoid]":
    """
    Saves data from a procedure to this file and returns PVoid.

    Creates a procedure that stores the output of the given procedure to this
    file. Unlike cache(), this returns PVoid, making it suitable for use in
    ProcedureSequence workflows where no return value is needed.

    The data is persisted at the end of procedure execution, after the user
    indicates they are done. If the user cancels early, no partial data is saved.

    Args:
        procedure: The procedure whose output should be saved to this file
        skip_if_exists: Whether to skip procedure if the file already exists

    Returns:
        A procedure that returns PVoid after saving is complete

    Example:
        ```python
        session = target.new_session("daily_capture")
        session.new_file("photo").save(CaptureImage(label="Daily Photo"))
        ```
    """
    ...

save_bundle(procedure, skip_if_exists=False)

Saves data from a procedure to this folder and returns PVoid.

Creates a procedure that stores the output of the given procedure to this folder. Unlike cache_bundle(), this returns PVoid, making it suitable for use in ProcedureSequence workflows where no return value is needed.

The data is persisted at the end of procedure execution, after the user indicates they are done. If the user cancels early, no partial data is saved.

Parameters:

Name	Type	Description	Default
procedure	Procedure[PSaveableBundle]	The procedure whose output should be saved to this folder	required
skip_if_exists	bool	Whether to skip procedure if the folder already exists	False

Returns:

Type	Description
Procedure[PVoid]	A procedure that returns PVoid after saving is complete

Example

session = target.new_session("daily_capture")
session.new_folder("photos").save_bundle(CaptureImageSequence(label="Daily Photos"))

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def save_bundle(
    self: "Procedure[PDirectory]",
    procedure: "Procedure[PSaveableBundle]",
    skip_if_exists: bool = False,
) -> "Procedure[PVoid]":
    """
    Saves data from a procedure to this folder and returns PVoid.

    Creates a procedure that stores the output of the given procedure to this
    folder. Unlike cache_bundle(), this returns PVoid, making it suitable for use in
    ProcedureSequence workflows where no return value is needed.

    The data is persisted at the end of procedure execution, after the user
    indicates they are done. If the user cancels early, no partial data is saved.

    Args:
        procedure: The procedure whose output should be saved to this folder
        skip_if_exists: Whether to skip procedure if the folder already exists

    Returns:
        A procedure that returns PVoid after saving is complete

    Example:
        ```python
        session = target.new_session("daily_capture")
        session.new_folder("photos").save_bundle(CaptureImageSequence(label="Daily Photos"))
        ```
    """
    ...

cache(procedure, skip_if_exists=False)

Caches data from a procedure to this file and returns the original data type.

Creates a procedure that stores the output of the given procedure to this file AND returns the data for continued use in the workflow. This is useful for persistent configuration data that you want to use in future sessions and also reference in conditional logic.

The data is persisted at the end of procedure execution. If skip_if_exists is True and the file exists, the existing data is preloaded for user editing.

Parameters:

Name	Type	Description	Default
procedure	Procedure[T2]	The procedure whose output should be cached to this file	required
skip_if_exists	bool	Whether to skip procedure if the file already exists	False

Returns:

Type	Description
Procedure[T2]	A procedure that returns the same type as the input procedure

Example

user_setting = target.new_file("setting").cache(
    UserInputString(label="Your preference"),
    skip_if_exists=True
)
# Now user_setting can be used in conditional logic

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def cache[T2: PSaveable](
    self: "Procedure[PFile]",
    procedure: "Procedure[T2]",
    skip_if_exists: bool = False,
) -> "Procedure[T2]":
    """
    Caches data from a procedure to this file and returns the original data type.

    Creates a procedure that stores the output of the given procedure to this
    file AND returns the data for continued use in the workflow. This is useful
    for persistent configuration data that you want to use in future sessions
    and also reference in conditional logic.

    The data is persisted at the end of procedure execution. If skip_if_exists
    is True and the file exists, the existing data is preloaded for user editing.

    Args:
        procedure: The procedure whose output should be cached to this file
        skip_if_exists: Whether to skip procedure if the file already exists

    Returns:
        A procedure that returns the same type as the input procedure

    Example:
        ```python
        user_setting = target.new_file("setting").cache(
            UserInputString(label="Your preference"),
            skip_if_exists=True
        )
        # Now user_setting can be used in conditional logic
        ```
    """
    ...

cache_bundle(procedure, skip_if_exists=False)

Caches data from a procedure to this folder and returns the original data type.

Creates a procedure that stores the output of the given procedure to this folder AND returns the data for continued use in the workflow. This is useful for persistent configuration data that you want to use in future sessions and also reference in conditional logic.

The data is persisted at the end of procedure execution. If skip_if_exists is True and the folder exists, the existing data is preloaded for user editing.

Parameters:

Name	Type	Description	Default
procedure	Procedure[T2]	The procedure whose output should be cached to this folder	required
skip_if_exists	bool	Whether to skip procedure if the folder already exists	False

Returns:

Type	Description
Procedure[T2]	A procedure that returns the same type as the input procedure

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def cache_bundle[T2: PSaveableBundle](
    self: "Procedure[PDirectory]",
    procedure: "Procedure[T2]",
    skip_if_exists: bool = False,
) -> "Procedure[T2]":
    """
    Caches data from a procedure to this folder and returns the original data type.

    Creates a procedure that stores the output of the given procedure to this
    folder AND returns the data for continued use in the workflow. This is useful
    for persistent configuration data that you want to use in future sessions
    and also reference in conditional logic.

    The data is persisted at the end of procedure execution. If skip_if_exists
    is True and the folder exists, the existing data is preloaded for user editing.

    Args:
        procedure: The procedure whose output should be cached to this folder
        skip_if_exists: Whether to skip procedure if the folder already exists

    Returns:
        A procedure that returns the same type as the input procedure
    """
    ...

load(or_else)

Loads existing data from this file or executes fallback procedure.

Creates a procedure that tries to load previously saved data from this file. If the file doesn't exist or has no data, it executes the fallback procedure instead. This is commonly used for reference-based workflows where you want to load existing references or create new ones.

Parameters:

Name	Type	Description	Default
or_else	Procedure[T2]	The fallback procedure to execute if no data exists in the file	required

Returns:

Type	Description
Procedure[T2]	A procedure that returns the loaded data or fallback result

Example

reference = target.new_file("reference_image").load(
    or_else=CaptureImage(label="Set baseline photo")
)
# Returns loaded image if exists, or prompts for new capture

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def load[T2: PSaveable](
    self: "Procedure[PFile]",
    or_else: "Procedure[T2]",
) -> "Procedure[T2]":
    """
    Loads existing data from this file or executes fallback procedure.

    Creates a procedure that tries to load previously saved data from this file.
    If the file doesn't exist or has no data, it executes the fallback procedure
    instead. This is commonly used for reference-based workflows where you want
    to load existing references or create new ones.

    Args:
        or_else: The fallback procedure to execute if no data exists in the file

    Returns:
        A procedure that returns the loaded data or fallback result

    Example:
        ```python
        reference = target.new_file("reference_image").load(
            or_else=CaptureImage(label="Set baseline photo")
        )
        # Returns loaded image if exists, or prompts for new capture
        ```
    """
    ...

load_bundle(or_else)

Shortcut for CacheProcedureToBundle node (load mode).

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def load_bundle[T2: PSaveableBundle](
    self: "Procedure[PDirectory]",
    or_else: "Procedure[T2]",
) -> "Procedure[T2]":
    """Shortcut for CacheProcedureToBundle node (load mode)."""
    ...

__or__(other)

Shortcut for BoolOr node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def __or__(
    self: "Procedure[PBool]", other: "Procedure[PBool]"
) -> "Procedure[PBool]":
    """Shortcut for BoolOr node."""
    ...

__and__(other)

Shortcut for BoolAnd node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def __and__(
    self: "Procedure[PBool]", other: "Procedure[PBool]"
) -> "Procedure[PBool]":
    """Shortcut for BoolAnd node."""
    ...

__invert__()

Shortcut for BoolNot node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def __invert__(self: "Procedure[PBool]") -> "Procedure[PBool]":
    """Shortcut for BoolNot node."""
    ...

__lt__(other)

Shortcut for NumberLessThan node.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def __lt__(
    self: "Procedure[PNumber]", other: "Procedure[PNumber]"
) -> "Procedure[PBool]":
    """Shortcut for NumberLessThan node."""
    ...

__gt__(other)

Shortcut for NumberLessThan node (with operands swapped).

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def __gt__(
    self: "Procedure[PNumber]", other: "Procedure[PNumber]"
) -> "Procedure[PBool]":
    """Shortcut for NumberLessThan node (with operands swapped)."""
    return other < self

__le__(other)

Shortcut for BoolNot(NumberLessThan) node combination.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def __le__(
    self: "Procedure[PNumber]", other: "Procedure[PNumber]"
) -> "Procedure[PBool]":
    """Shortcut for BoolNot(NumberLessThan) node combination."""
    return ~(other < self)

__ge__(other)

Shortcut for BoolNot(NumberLessThan) node combination (with operands swapped).

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def __ge__(
    self: "Procedure[PNumber]", other: "Procedure[PNumber]"
) -> "Procedure[PBool]":
    """Shortcut for BoolNot(NumberLessThan) node combination (with operands swapped)."""
    return ~(self < other)

forward_types(base_type, *argument_names)

Decorator to propagate the return type of a Procedure subclass based on the return_type of other Procedure instances passed as fields.

This decorator wraps the class's post_init method, inspects the specified dataclass fields (by name), and ensures all of them are Procedure instances returning the same PType. The subclass's _return_type attribute is then set to this common return type.

Parameters:

Name	Type	Description	Default
*argument_names	str	Names of the dataclass fields whose return types should be forwarded.	()

Returns:

Type	Description
Callable[[C], C]	The class decorator that updates post_init on the target class.

Source code in capturegraph-lib/capturegraph/procedures/procedure.py

def forward_types[C: type[Procedure[PType]]](
    base_type: type[PType], *argument_names: str
) -> Callable[[C], C]:
    """
    Decorator to propagate the return type of a Procedure subclass based
    on the return_type of other Procedure instances passed as fields.

    This decorator wraps the class's __post_init__ method, inspects
    the specified dataclass fields (by name), and ensures all of them
    are Procedure instances returning the same PType. The subclass's
    _return_type attribute is then set to this common return type.

    Args:
        *argument_names: Names of the dataclass fields whose return types
                         should be forwarded.

    Returns:
        The class decorator that updates __post_init__ on the target class.
    """

    #: private
    def decorator(cls: C) -> C:
        original_post_init = cls.__post_init__

        def __post_init__(self) -> None:
            forwarded_type = None
            for name, value in self._context.items():
                if name not in argument_names:
                    continue
                elif not isinstance(value, Procedure):
                    raise TypeError(
                        f"Field '{name}' in '{self.__class__.__name__}' "
                        "must be a Procedure instance."
                    )
                elif forwarded_type is None:
                    if not issubclass(value._return_type, base_type):
                        raise TypeError(
                            f"Input '{name}' for node '{self}' must return '{base_type}' "
                            f"but returns '{value._return_type}' instead."
                        )

                    forwarded_type = value._return_type
                elif forwarded_type != value._return_type:
                    raise TypeError(
                        f"Inconsistent return types in '{name}': "
                        f"expected {forwarded_type}, got {value._return_type}"
                    )

            if forwarded_type is not None:
                object.__setattr__(self, "_return_type", forwarded_type)

            original_post_init(self)

        setattr(cls, "__post_init__", __post_init__)
        return cls

    return decorator