cmflib.cmf

This class provides methods to log metadata for distributed AI pipelines. The class instance creates an ML metadata store to store the metadata. It creates a driver to store nodes and its relationships to neo4j. The user has to provide the name of the pipeline, that needs to be recorded with CMF.

cmflib.cmf.Cmf(
    filepath="mlmd",
    pipeline_name="test_pipeline",
    custom_properties={"owner": "user_a"},
    graph=False
)

Args: filepath: Path to the sqlite file to store the metadata pipeline_name: Name to uniquely identify the pipeline. Note that name is the unique identifier for a pipeline. If a pipeline already exist with the same name, the existing pipeline object is reused. custom_properties: Additional properties of the pipeline that needs to be stored. graph: If set to true, the libray also stores the relationships in the provided graph database. The following variables should be set: neo4j_uri (graph server URI), neo4j_user (user name) and neo4j_password (user password), e.g.:

cmf init local --path /home/user/local-storage --git-remote-url https://github.com/XXX/exprepo.git --neo4j-user neo4j --neo4j-password neo4j
                      --neo4j-uri bolt://localhost:7687

Source code in cmflib/cmf.py

    def __init__(
        self,
        filepath: str = "mlmd",
        pipeline_name: str = "",
        custom_properties: t.Optional[t.Dict] = None,
        graph: bool = False,
        is_server: bool = False,
    ):
        #path to directory
        self.cmf_init_path = filepath.rsplit("/",1)[0] \
				 if len(filepath.rsplit("/",1)) > 1 \
					else  os.getcwd()

        logging_dir = change_dir(self.cmf_init_path)
        if is_server is False:
            Cmf.__prechecks()
        if custom_properties is None:
            custom_properties = {}
        if not pipeline_name:
            # assign folder name as pipeline name 
            cur_folder = os.path.basename(os.getcwd())
            pipeline_name = cur_folder
        config = mlpb.ConnectionConfig()
        config.sqlite.filename_uri = filepath
        self.store = metadata_store.MetadataStore(config)
        self.filepath = filepath
        self.child_context = None
        self.execution = None
        self.execution_name = ""
        self.execution_command = ""
        self.metrics = {}
        self.input_artifacts = []
        self.execution_label_props = {}
        self.graph = graph
        #last token in filepath
        self.branch_name = filepath.rsplit("/", 1)[-1]

        if is_server is False:
            git_checkout_new_branch(self.branch_name)
        self.parent_context = get_or_create_parent_context(
            store=self.store,
            pipeline=pipeline_name,
            custom_properties=custom_properties,
        )
        if is_server:
            Cmf.__get_neo4j_server_config()
        if graph is True:
            Cmf.__load_neo4j_params()
            self.driver = graph_wrapper.GraphDriver(
                Cmf.__neo4j_uri, Cmf.__neo4j_user, Cmf.__neo4j_password
            )
            self.driver.create_pipeline_node(
                pipeline_name, self.parent_context.id, custom_properties
            )
        os.chdir(logging_dir)

create_context(pipeline_stage, custom_properties=None)

Create's a context(stage). Every call creates a unique pipeline stage. Updates Pipeline_stage name. Example:

#Create context
# Import CMF
from cmflib.cmf import Cmf
from ml_metadata.proto import metadata_store_pb2 as mlpb
# Create CMF logger
cmf = Cmf(filepath="mlmd", pipeline_name="test_pipeline")
# Create context
context: mlmd.proto.Context = cmf.create_context(
    pipeline_stage="prepare",
    custom_properties ={"user-metadata1": "metadata_value"}
)

Args: Pipeline_stage: Name of the Stage. custom_properties: Developers can provide key value pairs with additional properties of the execution that need to be stored. Returns: Context object from ML Metadata library associated with the new context for this stage.

Source code in cmflib/cmf.py

def create_context(
    self, pipeline_stage: str, custom_properties: t.Optional[t.Dict] = None
) -> mlpb.Context:
    """Create's a  context(stage).
    Every call creates a unique pipeline stage.
    Updates Pipeline_stage name.
    Example:
        ```python
        #Create context
        # Import CMF
        from cmflib.cmf import Cmf
        from ml_metadata.proto import metadata_store_pb2 as mlpb
        # Create CMF logger
        cmf = Cmf(filepath="mlmd", pipeline_name="test_pipeline")
        # Create context
        context: mlmd.proto.Context = cmf.create_context(
            pipeline_stage="prepare",
            custom_properties ={"user-metadata1": "metadata_value"}
        )

        ```
        Args:
            Pipeline_stage: Name of the Stage.
            custom_properties: Developers can provide key value pairs with additional properties of the execution that
                need to be stored.
        Returns:
            Context object from ML Metadata library associated with the new context for this stage.
    """
    custom_props = {} if custom_properties is None else custom_properties
    pipeline_stage = self.parent_context.name + "/" + pipeline_stage
    ctx = get_or_create_run_context(
        self.store, pipeline_stage, custom_props)
    self.child_context = ctx
    associate_child_to_parent_context(
        store=self.store, parent_context=self.parent_context, child_context=ctx
    )
    if self.graph:
        self.driver.create_stage_node(
            pipeline_stage, self.parent_context, ctx.id, custom_props
        )
    return ctx

create_execution(execution_type, custom_properties=None, cmd=None, create_new_execution=True)

Create execution. Every call creates a unique execution. Execution can only be created within a context, so create_context must be called first. Example:

# Import CMF
from cmflib.cmf import Cmf
from ml_metadata.proto import metadata_store_pb2 as mlpb
# Create CMF logger
cmf = Cmf(filepath="mlmd", pipeline_name="test_pipeline")
# Create or reuse context for this stage
context: mlmd.proto.Context = cmf.create_context(
    pipeline_stage="prepare",
    custom_properties ={"user-metadata1": "metadata_value"}
)
# Create a new execution for this stage run
execution: mlmd.proto.Execution = cmf.create_execution(
    execution_type="Prepare",
    custom_properties = {"split": split, "seed": seed}
)

Args: execution_type: Type of the execution.(when create_new_execution is False, this is the name of execution) custom_properties: Developers can provide key value pairs with additional properties of the execution that need to be stored.

cmd: command used to run this execution.

create_new_execution:bool = True, This can be used by advanced users to re-use executions
    This is applicable, when working with framework code like mmdet, pytorch lightning etc, where the
    custom call-backs are used to log metrics.
    if create_new_execution is True(Default), execution_type parameter will be used as the name of the execution type.
    if create_new_execution is False, if existing execution exist with the same name as execution_type.
    it will be reused.
    Only executions created with  create_new_execution as False will have "name" as a property.

Returns:

Type	Description
Execution	Execution object from ML Metadata library associated with the new execution for this stage.

Source code in cmflib/cmf.py

def create_execution(
    self,
    execution_type: str,
    custom_properties: t.Optional[t.Dict] = None,
    cmd: str = None,
    create_new_execution: bool = True,
) -> mlpb.Execution:
    """Create execution.
    Every call creates a unique execution. Execution can only be created within a context, so
    [create_context][cmflib.cmf.Cmf.create_context] must be called first.
    Example:
        ```python
        # Import CMF
        from cmflib.cmf import Cmf
        from ml_metadata.proto import metadata_store_pb2 as mlpb
        # Create CMF logger
        cmf = Cmf(filepath="mlmd", pipeline_name="test_pipeline")
        # Create or reuse context for this stage
        context: mlmd.proto.Context = cmf.create_context(
            pipeline_stage="prepare",
            custom_properties ={"user-metadata1": "metadata_value"}
        )
        # Create a new execution for this stage run
        execution: mlmd.proto.Execution = cmf.create_execution(
            execution_type="Prepare",
            custom_properties = {"split": split, "seed": seed}
        )
        ```
    Args:
        execution_type: Type of the execution.(when create_new_execution is False, this is the name of execution)
        custom_properties: Developers can provide key value pairs with additional properties of the execution that
            need to be stored.

        cmd: command used to run this execution.

        create_new_execution:bool = True, This can be used by advanced users to re-use executions
            This is applicable, when working with framework code like mmdet, pytorch lightning etc, where the
            custom call-backs are used to log metrics.
            if create_new_execution is True(Default), execution_type parameter will be used as the name of the execution type.
            if create_new_execution is False, if existing execution exist with the same name as execution_type.
            it will be reused.
            Only executions created with  create_new_execution as False will have "name" as a property.


    Returns:
        Execution object from ML Metadata library associated with the new execution for this stage.
    """
    logging_dir = change_dir(self.cmf_init_path)
    # Assigning current file name as stage and execution name
    current_script = sys.argv[0]
    file_name = os.path.basename(current_script)
    name_without_extension = os.path.splitext(file_name)[0]
    # create context if not already created
    if not self.child_context:
        self.create_context(pipeline_stage=name_without_extension)
        assert self.child_context is not None, f"Failed to create context for {self.pipeline_name}!!"

    # Initializing the execution related fields

    self.metrics = {}
    self.input_artifacts = []
    self.execution_label_props = {}
    custom_props = {} if custom_properties is None else custom_properties
    git_repo = git_get_repo()
    git_start_commit = git_get_commit()
    cmd = str(sys.argv) if cmd is None else cmd
    python_env=get_python_env()
    self.execution = create_new_execution_in_existing_run_context(
        store=self.store,
        # Type field when re-using executions
        execution_type_name=self.child_context.name,
        execution_name=execution_type, 
        #Name field if we are re-using executions
        #Type field , if creating new executions always 
        context_id=self.child_context.id,
        execution=cmd,
        pipeline_id=self.parent_context.id,
        pipeline_type=self.parent_context.name,
        git_repo=git_repo,
        git_start_commit=git_start_commit,
        python_env=python_env,
        custom_properties=custom_props,
        create_new_execution=create_new_execution,
    )
    uuids = self.execution.properties["Execution_uuid"].string_value
    if uuids:
        self.execution.properties["Execution_uuid"].string_value = uuids+","+str(uuid.uuid1())
    else:
        self.execution.properties["Execution_uuid"].string_value = str(uuid.uuid1())            
    self.store.put_executions([self.execution])
    self.execution_name = str(self.execution.id) + "," + execution_type
    self.execution_command = cmd
    for k, v in custom_props.items():
        k = re.sub("-", "_", k)
        self.execution_label_props[k] = v
    self.execution_label_props["Execution_Name"] = (
        execution_type + ":" + str(self.execution.id)
    )

    self.execution_label_props["execution_command"] = cmd
    if self.graph:
        self.driver.create_execution_node(
        self.execution_name,
        self.child_context.id,
        self.parent_context,
        cmd,
        self.execution.id,
        custom_props,
    )
    os.chdir(logging_dir)
    return self.execution

update_execution(execution_id, custom_properties=None)

Updates an existing execution. The custom properties can be updated after creation of the execution. The new custom properties is merged with earlier custom properties. Example

# Import CMF
from cmflib.cmf import Cmf
from ml_metadata.proto import metadata_store_pb2 as mlpb
# Create CMF logger
cmf = Cmf(filepath="mlmd", pipeline_name="test_pipeline")
# Update a execution
execution: mlmd.proto.Execution = cmf.update_execution(
    execution_id=8,
    custom_properties = {"split": split, "seed": seed}
)

Args: execution_id: id of the execution. custom_properties: Developers can provide key value pairs with additional properties of the execution that need to be updated. Returns: Execution object from ML Metadata library associated with the updated execution for this stage.

Source code in cmflib/cmf.py

def update_execution(
    self, execution_id: int, custom_properties: t.Optional[t.Dict] = None
):
    """Updates an existing execution.
    The custom properties can be updated after creation of the execution.
    The new custom properties is merged with earlier custom properties.
    Example
        ```python
        # Import CMF
        from cmflib.cmf import Cmf
        from ml_metadata.proto import metadata_store_pb2 as mlpb
        # Create CMF logger
        cmf = Cmf(filepath="mlmd", pipeline_name="test_pipeline")
        # Update a execution
        execution: mlmd.proto.Execution = cmf.update_execution(
            execution_id=8,
            custom_properties = {"split": split, "seed": seed}
        )
        ```
        Args:
            execution_id: id of the execution.
            custom_properties: Developers can provide key value pairs with additional properties of the execution that
            need to be updated.
        Returns:
            Execution object from ML Metadata library associated with the updated execution for this stage.
    """
    self.execution = self.store.get_executions_by_id([execution_id])[0]
    if self.execution is None:
        print("Error - no execution id")
        return
    execution_type = self.store.get_execution_types_by_id([self.execution.type_id])[
        0
    ]

    if custom_properties:
        for key, value in custom_properties.items():
            if isinstance(value, int):
                self.execution.custom_properties[key].int_value = value
            else:
                self.execution.custom_properties[key].string_value = str(
                    value)
    self.store.put_executions([self.execution])
    c_props = {}
    for k, v in self.execution.custom_properties.items():
        key = re.sub("-", "_", k)
        val_type = str(v).split(":", maxsplit=1)[0]
        if val_type == "string_value":
            val = self.execution.custom_properties[k].string_value
        else:
            val = str(v).split(":")[1]
        # The properties value are stored in the format type:value hence,
        # taking only value
        self.execution_label_props[key] = val
        c_props[key] = val
    self.execution_name = str(self.execution.id) + \
        "," + execution_type.name
    self.execution_command = self.execution.properties["Execution"]
    self.execution_label_props["Execution_Name"] = (
        execution_type.name + ":" + str(self.execution.id)
    )
    self.execution_label_props["execution_command"] = self.execution.properties[
        "Execution"
    ].string_value
    if self.graph:
        self.driver.create_execution_node(
            self.execution_name,
            self.child_context.id,
            self.parent_context,
            self.execution.properties["Execution"].string_value,
            self.execution.id,
            c_props,
        )
    return self.execution

log_dataset(url, event, custom_properties=None, external=False)

Logs a dataset as artifact. This call adds the dataset to dvc. The dvc metadata file created (.dvc) will be added to git and committed. The version of the dataset is automatically obtained from the versioning software(DVC) and tracked as a metadata. Example:

artifact: mlmd.proto.Artifact = cmf.log_dataset(
    url="/repo/data.xml",
    event="input",
    custom_properties={"source":"kaggle"}
)

Args: url: The path to the dataset. event: Takes arguments INPUT OR OUTPUT. custom_properties: Dataset properties (key/value pairs). Returns: Artifact object from ML Metadata library associated with the new dataset artifact.

Source code in cmflib/cmf.py

def log_dataset(
    self,
    url: str,
    event: str,
    custom_properties: t.Optional[t.Dict] = None,
    external: bool = False,
) -> mlpb.Artifact:
    """Logs a dataset as artifact.
    This call adds the dataset to dvc. The dvc metadata file created (.dvc) will be added to git and committed. The
    version of the  dataset is automatically obtained from the versioning software(DVC) and tracked as a metadata.
    Example:
        ```python
        artifact: mlmd.proto.Artifact = cmf.log_dataset(
            url="/repo/data.xml",
            event="input",
            custom_properties={"source":"kaggle"}
        )
        ```
    Args:
         url: The path to the dataset.
         event: Takes arguments `INPUT` OR `OUTPUT`.
         custom_properties: Dataset properties (key/value pairs).
    Returns:
        Artifact object from ML Metadata library associated with the new dataset artifact.
    """
    logging_dir = change_dir(self.cmf_init_path)
    # Assigning current file name as stage and execution name
    current_script = sys.argv[0]
    file_name = os.path.basename(current_script)
    name_without_extension = os.path.splitext(file_name)[0]
    # create context if not already created
    if not self.child_context:
        self.create_context(pipeline_stage=name_without_extension)
        assert self.child_context is not None, f"Failed to create context for {self.pipeline_name}!!"

    # create execution if not already created
    if not self.execution:
        self.create_execution(execution_type=name_without_extension)
        assert self.execution is not None, f"Failed to create execution for {self.pipeline_name}!!"

            ### To Do : Technical Debt. 
    # If the dataset already exist , then we just link the existing dataset to the execution
    # We do not update the dataset properties . 
    # We need to append the new properties to the existing dataset properties
    custom_props = {} if custom_properties is None else custom_properties
    git_repo = git_get_repo()
    name = re.split("/", url)[-1]
    event_type = mlpb.Event.Type.OUTPUT
    existing_artifact = []
    if event.lower() == "input":
        event_type = mlpb.Event.Type.INPUT

    commit_output(url, self.execution.id)
    c_hash = dvc_get_hash(url)

    if c_hash == "":
        print("Error in getting the dvc hash,return without logging")
        return

    dataset_commit = c_hash
    dvc_url = dvc_get_url(url)
    dvc_url_with_pipeline = f"{self.parent_context.name}:{dvc_url}"
    url = url + ":" + c_hash
    if c_hash and c_hash.strip:
        existing_artifact.extend(self.store.get_artifacts_by_uri(c_hash))

    # To Do - What happens when uri is the same but names are different
    if existing_artifact and len(existing_artifact) != 0:
        existing_artifact = existing_artifact[0]

        # Quick fix- Updating only the name
        if custom_properties is not None:
            self.update_existing_artifact(
                existing_artifact, custom_properties)
        uri = c_hash
        # update url for existing artifact
        self.update_dataset_url(existing_artifact, dvc_url_with_pipeline)
        artifact = link_execution_to_artifact(
            store=self.store,
            execution_id=self.execution.id,
            uri=uri,
            input_name=url,
            event_type=event_type,
        )
    else:
        # if((existing_artifact and len(existing_artifact )!= 0) and c_hash != ""):
        #   url = url + ":" + str(self.execution.id)
        uri = c_hash if c_hash and c_hash.strip() else str(uuid.uuid1())
        artifact = create_new_artifact_event_and_attribution(
            store=self.store,
            execution_id=self.execution.id,
            context_id=self.child_context.id,
            uri=uri,
            name=url,
            type_name="Dataset",
            event_type=event_type,
            properties={
                "git_repo": str(git_repo),
                # passing c_hash value to commit
                "Commit": str(dataset_commit),
                "url": str(dvc_url_with_pipeline),
            },
            artifact_type_properties={
                "git_repo": mlpb.STRING,
                "Commit": mlpb.STRING,
                "url": mlpb.STRING,
            },
            custom_properties=custom_props,
            milliseconds_since_epoch=int(time.time() * 1000),
        )
    custom_props["git_repo"] = git_repo
    custom_props["Commit"] = dataset_commit
    self.execution_label_props["git_repo"] = git_repo
    self.execution_label_props["Commit"] = dataset_commit

    if self.graph:
        self.driver.create_dataset_node(
            name,
            url,
            uri,
            event,
            self.execution.id,
            self.parent_context,
            custom_props,
        )
        if event.lower() == "input":
            self.input_artifacts.append(
                {
                    "Name": name,
                    "Path": url,
                    "URI": uri,
                    "Event": event.lower(),
                    "Execution_Name": self.execution_name,
                    "Type": "Dataset",
                    "Execution_Command": self.execution_command,
                    "Pipeline_Id": self.parent_context.id,
                    "Pipeline_Name": self.parent_context.name,
                }
            )
            self.driver.create_execution_links(uri, name, "Dataset")
        else:
            child_artifact = {
                "Name": name,
                "Path": url,
                "URI": uri,
                "Event": event.lower(),
                "Execution_Name": self.execution_name,
                "Type": "Dataset",
                "Execution_Command": self.execution_command,
                "Pipeline_Id": self.parent_context.id,
                "Pipeline_Name": self.parent_context.name,
            }
            self.driver.create_artifact_relationships(
                self.input_artifacts, child_artifact, self.execution_label_props
            )
    os.chdir(logging_dir)
    return artifact

log_model(path, event, model_framework='Default', model_type='Default', model_name='Default', custom_properties=None)

Logs a model. The model is added to dvc and the metadata file (.dvc) gets committed to git. Example:

artifact: mlmd.proto.Artifact= cmf.log_model(
    path="path/to/model.pkl",
    event="output",
    model_framework="SKlearn",
    model_type="RandomForestClassifier",
    model_name="RandomForestClassifier:default"
)

Args: path: Path to the model file. event: Takes arguments INPUT OR OUTPUT. model_framework: Framework used to create the model. model_type: Type of model algorithm used. model_name: Name of the algorithm used. custom_properties: The model properties. Returns: Artifact object from ML Metadata library associated with the new model artifact.

Source code in cmflib/cmf.py

def log_model(
    self,
    path: str,
    event: str,
    model_framework: str = "Default",
    model_type: str = "Default",
    model_name: str = "Default",
    custom_properties: t.Optional[t.Dict] = None,
) -> mlpb.Artifact:
    """Logs a model.
    The model is added to dvc and the metadata file (.dvc) gets committed to git.
    Example:
        ```python
        artifact: mlmd.proto.Artifact= cmf.log_model(
            path="path/to/model.pkl",
            event="output",
            model_framework="SKlearn",
            model_type="RandomForestClassifier",
            model_name="RandomForestClassifier:default"
        )
        ```
    Args:
        path: Path to the model file.
        event: Takes arguments `INPUT` OR `OUTPUT`.
        model_framework: Framework used to create the model.
        model_type: Type of model algorithm used.
        model_name: Name of the algorithm used.
        custom_properties: The model properties.
    Returns:
        Artifact object from ML Metadata library associated with the new model artifact.
    """

    logging_dir = change_dir(self.cmf_init_path)
    # Assigning current file name as stage and execution name
    current_script = sys.argv[0]
    file_name = os.path.basename(current_script)
    name_without_extension = os.path.splitext(file_name)[0]
    # create context if not already created
    if not self.child_context:
        self.create_context(pipeline_stage=name_without_extension)
        assert self.child_context is not None, f"Failed to create context for {self.pipeline_name}!!"

    # create execution if not already created
    if not self.execution:
        self.create_execution(execution_type=name_without_extension)
        assert self.execution is not None, f"Failed to create execution for {self.pipeline_name}!!"


    # To Do : Technical Debt. 
    # If the model already exist , then we just link the existing model to the execution
    # We do not update the model properties . 
    # We need to append the new properties to the existing model properties
    if custom_properties is None:
        custom_properties = {}
    custom_props = {} if custom_properties is None else custom_properties
    # name = re.split('/', path)[-1]
    event_type = mlpb.Event.Type.OUTPUT
    existing_artifact = []
    if event.lower() == "input":
        event_type = mlpb.Event.Type.INPUT

    commit_output(path, self.execution.id)
    c_hash = dvc_get_hash(path)

    if c_hash == "":
        print("Error in getting the dvc hash,return without logging")
        return

    model_commit = c_hash

    # If connecting to an existing artifact - The name of the artifact is
    # used as path/steps/key
    model_uri = path + ":" + c_hash
    dvc_url = dvc_get_url(path, False)
    url = dvc_url
    url_with_pipeline = f"{self.parent_context.name}:{url}"
    uri = ""
    if c_hash and c_hash.strip():
        uri = c_hash.strip()
        existing_artifact.extend(self.store.get_artifacts_by_uri(uri))
    else:
        raise RuntimeError("Model commit failed, Model uri empty")

    if (
        existing_artifact
        and len(existing_artifact) != 0
    ):
        # update url for existing artifact
        existing_artifact = self.update_model_url(
            existing_artifact, url_with_pipeline
        )
        artifact = link_execution_to_artifact(
            store=self.store,
            execution_id=self.execution.id,
            uri=c_hash,
            input_name=model_uri,
            event_type=event_type,
        )
        model_uri =  model_uri + ":" + str(self.execution.id)
    else:
        uri = c_hash if c_hash and c_hash.strip() else str(uuid.uuid1())
        model_uri = model_uri + ":" + str(self.execution.id)
        artifact = create_new_artifact_event_and_attribution(
            store=self.store,
            execution_id=self.execution.id,
            context_id=self.child_context.id,
            uri=uri,
            name=model_uri,
            type_name="Model",
            event_type=event_type,
            properties={
                "model_framework": str(model_framework),
                "model_type": str(model_type),
                "model_name": str(model_name),
                # passing c_hash value to commit
                "Commit": str(model_commit),
                "url": str(url_with_pipeline),
            },
            artifact_type_properties={
                "model_framework": mlpb.STRING,
                "model_type": mlpb.STRING,
                "model_name": mlpb.STRING,
                "Commit": mlpb.STRING,
                "url": mlpb.STRING,
            },
            custom_properties=custom_props,
            milliseconds_since_epoch=int(time.time() * 1000),
        )
    # custom_properties["Commit"] = model_commit
    self.execution_label_props["Commit"] = model_commit
    #To DO model nodes should be similar to dataset nodes when we create neo4j
    if self.graph:
        self.driver.create_model_node(
            model_uri,
            uri,
            event,
            self.execution.id,
            self.parent_context,
            custom_props,
        )
        if event.lower() == "input":
            self.input_artifacts.append(
                {
                    "Name": model_uri,
                    "URI": uri,
                    "Event": event.lower(),
                    "Execution_Name": self.execution_name,
                    "Type": "Model",
                    "Execution_Command": self.execution_command,
                    "Pipeline_Id": self.parent_context.id,
                    "Pipeline_Name": self.parent_context.name,
                }
            )
            self.driver.create_execution_links(uri, model_uri, "Model")
        else:
            child_artifact = {
                "Name": model_uri,
                "URI": uri,
                "Event": event.lower(),
                "Execution_Name": self.execution_name,
                "Type": "Model",
                "Execution_Command": self.execution_command,
                "Pipeline_Id": self.parent_context.id,
                "Pipeline_Name": self.parent_context.name,
            }

            self.driver.create_artifact_relationships(
                self.input_artifacts, child_artifact, self.execution_label_props
            )
    os.chdir(logging_dir)
    return artifact

log_execution_metrics(metrics_name, custom_properties=None)

Log the metadata associated with the execution (coarse-grained tracking). It is stored as a metrics artifact. This does not have a backing physical file, unlike other artifacts that we have. Example:

exec_metrics: mlpb.Artifact = cmf.log_execution_metrics(
    metrics_name="Training_Metrics",
    {"auc": auc, "loss": loss}
)

Args: metrics_name: Name to identify the metrics. custom_properties: Dictionary with metric values. Returns: Artifact object from ML Metadata library associated with the new coarse-grained metrics artifact.

Source code in cmflib/cmf.py

def log_execution_metrics(
    self, metrics_name: str, custom_properties: t.Optional[t.Dict] = None
) -> mlpb.Artifact:
    """Log the metadata associated with the execution (coarse-grained tracking).
    It is stored as a metrics artifact. This does not have a backing physical file, unlike other artifacts that we
    have.
    Example:
        ```python
        exec_metrics: mlpb.Artifact = cmf.log_execution_metrics(
            metrics_name="Training_Metrics",
            {"auc": auc, "loss": loss}
        )
        ```
    Args:
        metrics_name: Name to identify the metrics.
        custom_properties: Dictionary with metric values.
    Returns:
          Artifact object from ML Metadata library associated with the new coarse-grained metrics artifact.
    """
    logging_dir = change_dir(self.cmf_init_path)
    # Assigning current file name as stage and execution name
    current_script = sys.argv[0]
    file_name = os.path.basename(current_script)
    name_without_extension = os.path.splitext(file_name)[0]
    # create context if not already created
    if not self.child_context:
        self.create_context(pipeline_stage=name_without_extension)
        assert self.child_context is not None, f"Failed to create context for {self.pipeline_name}!!"

    # create execution if not already created
    if not self.execution:
        self.create_execution(execution_type=name_without_extension)
        assert self.execution is not None, f"Failed to create execution for {self.pipeline_name}!!"

    custom_props = {} if custom_properties is None else custom_properties
    uri = str(uuid.uuid1())
    metrics_name = metrics_name + ":" + uri + ":" + str(self.execution.id)
    metrics = create_new_artifact_event_and_attribution(
        store=self.store,
        execution_id=self.execution.id,
        context_id=self.child_context.id,
        uri=uri,
        name=metrics_name,
        type_name="Metrics",
        event_type=mlpb.Event.Type.OUTPUT,
        properties={"metrics_name": metrics_name},
        artifact_type_properties={"metrics_name": mlpb.STRING},
        custom_properties=custom_props,
        milliseconds_since_epoch=int(time.time() * 1000),
    )
    if self.graph:
        # To do create execution_links
        self.driver.create_metrics_node(
            metrics_name,
            uri,
            "output",
            self.execution.id,
            self.parent_context,
            custom_props,
        )
        child_artifact = {
            "Name": metrics_name,
            "URI": uri,
            "Event": "output",
            "Execution_Name": self.execution_name,
            "Type": "Metrics",
            "Execution_Command": self.execution_command,
            "Pipeline_Id": self.parent_context.id,
            "Pipeline_Name": self.parent_context.name,
        }
        self.driver.create_artifact_relationships(
            self.input_artifacts, child_artifact, self.execution_label_props
        )
    os.chdir(logging_dir)
    return metrics

log_metric(metrics_name, custom_properties=None)

Stores the fine-grained (per step or per epoch) metrics to memory. The metrics provided are stored in a parquet file. The commit_metrics call add the parquet file in the version control framework. The metrics written in the parquet file can be retrieved using the read_metrics call. Example:

# Can be called at every epoch or every step in the training. This is logged to a parquet file and committed
# at the commit stage.
# Inside training loop
while True:
     cmf.log_metric("training_metrics", {"train_loss": train_loss})
cmf.commit_metrics("training_metrics")

Args: metrics_name: Name to identify the metrics. custom_properties: Dictionary with metrics.

Source code in cmflib/cmf.py

def log_metric(
    self, metrics_name: str, custom_properties: t.Optional[t.Dict] = None
) -> None:
    """Stores the fine-grained (per step or per epoch) metrics to memory.
    The metrics provided are stored in a parquet file. The `commit_metrics` call add the parquet file in the version
    control framework. The metrics written in the parquet file can be retrieved using the `read_metrics` call.
    Example:
        ```python
        # Can be called at every epoch or every step in the training. This is logged to a parquet file and committed
        # at the commit stage.
        # Inside training loop
        while True:
             cmf.log_metric("training_metrics", {"train_loss": train_loss})
        cmf.commit_metrics("training_metrics")
        ```
    Args:
        metrics_name: Name to identify the metrics.
        custom_properties: Dictionary with metrics.
    """
    if metrics_name in self.metrics:
        key = max((self.metrics[metrics_name]).keys()) + 1
        self.metrics[metrics_name][key] = custom_properties
    else:
        self.metrics[metrics_name] = {}
        self.metrics[metrics_name][1] = custom_properties

create_dataslice(name)

Creates a dataslice object. Once created, users can add data instances to this data slice with add_data method. Users are also responsible for committing data slices by calling the commit method. Example:

dataslice = cmf.create_dataslice("slice-a")

Args: name: Name to identify the dataslice.

Returns:

Type	Description
DataSlice	Instance of a newly created [DataSlice][cmflib.cmf.Cmf.DataSlice].

Source code in cmflib/cmf.py

def create_dataslice(self, name: str) -> "Cmf.DataSlice":
    """Creates a dataslice object.
    Once created, users can add data instances to this data slice with [add_data][cmflib.cmf.Cmf.DataSlice.add_data]
    method. Users are also responsible for committing data slices by calling the
    [commit][cmflib.cmf.Cmf.DataSlice.commit] method.
    Example:
        ```python
        dataslice = cmf.create_dataslice("slice-a")
        ```
    Args:
        name: Name to identify the dataslice.

    Returns:
        Instance of a newly created [DataSlice][cmflib.cmf.Cmf.DataSlice].
    """
    return Cmf.DataSlice(name, self)

update_dataslice(name, record, custom_properties)

Updates a dataslice record in a Parquet file with the provided custom properties. Example:

   dataslice=cmf.update_dataslice("dataslice_file.parquet", "record_id", 
   {"key1": "updated_value"})

Args: name: Name of the Parquet file. record: Identifier of the dataslice record to be updated. custom_properties: Dictionary containing custom properties to update.

Returns:

Type	Description
	None

Source code in cmflib/cmf.py

def update_dataslice(self, name: str, record: str, custom_properties: t.Dict):
    """Updates a dataslice record in a Parquet file with the provided custom properties.
    Example:
    ```python
       dataslice=cmf.update_dataslice("dataslice_file.parquet", "record_id", 
       {"key1": "updated_value"})
    ```
    Args:
       name: Name of the Parquet file.
       record: Identifier of the dataslice record to be updated.
       custom_properties: Dictionary containing custom properties to update.

    Returns:
       None
    """
    directory_path = os.path.join(self.ARTIFACTS_PATH, self.execution.properties["Execution_uuid"].string_value.split(',')[0], self.DATASLICE_PATH)
    name = os.path.join(directory_path, name)
    df = pd.read_parquet(name)
    temp_dict = df.to_dict("index")
    temp_dict[record].update(custom_properties)
    dataslice_df = pd.DataFrame.from_dict(temp_dict, orient="index")
    dataslice_df.index.names = ["Path"]
    dataslice_df.to_parquet(name)

This module contains all the public API for CMF

cmf_init_show()

Initializes and shows details of the CMF command. Example:

     result = cmf_init_show() 

Returns: Output from the _cmf_cmd_init function.

Source code in cmflib/cmf.py

def cmf_init_show():
    """ Initializes and shows details of the CMF command. 
    Example: 
    ```python 
         result = cmf_init_show() 
    ``` 
    Returns: 
       Output from the _cmf_cmd_init function. 
    """

    output=_cmf_cmd_init()
    return output

cmf_init(type='', path='', git_remote_url='', cmf_server_url='', neo4j_user='', neo4j_password='', neo4j_uri='', url='', endpoint_url='', access_key_id='', secret_key='', session_token='', user='', password='', port=0, osdf_path='', osdf_cache='', key_id='', key_path='', key_issuer='')

Initializes the CMF configuration based on the provided parameters. Example:

   cmf_init( type="local", 
             path="/path/to/re",
             git_remote_url="git@github.com:user/repo.git",
             cmf_server_url="http://cmf-server"
             neo4j_user", 
             neo4j_password="password",
             neo4j_uri="bolt://localhost:76"
           )

Args: type: Type of repository ("local", "minioS3", "amazonS3", "sshremote") path: Path for the local repository. git_remote_url: Git remote URL for version control. cmf_server_url: CMF server URL. neo4j_user: Neo4j database username. neo4j_password: Neo4j database password. neo4j_uri: Neo4j database URI. url: URL for MinioS3 or AmazonS3. endpoint_url: Endpoint URL for MinioS3. access_key_id: Access key ID for MinioS3 or AmazonS3. secret_key: Secret key for MinioS3 or AmazonS3. session_token: Session token for AmazonS3. user: SSH remote username. password: SSH remote password. port: SSH remote port. osdf_path: OSDF Origin Path. osdf_cache: OSDF Cache Path (Optional). key_id: OSDF Key ID. key_path: OSDF Private Key Path. key_issuer: OSDF Key Issuer URL. Returns: Output based on the initialized repository type.

Source code in cmflib/cmf.py

def cmf_init(type: str = "",
        path: str = "",
        git_remote_url: str = "",
        cmf_server_url: str = "",
        neo4j_user: str = "",
        neo4j_password: str = "",
        neo4j_uri: str = "",
        url: str = "",
        endpoint_url: str = "",
        access_key_id: str = "",
        secret_key: str = "",
        session_token: str = "",
        user: str = "",
        password: str = "",
        port: int = 0,
        osdf_path: str = "",
        osdf_cache: str = "",
        key_id: str = "",
        key_path: str = "",
        key_issuer: str = "",
         ):

    """ Initializes the CMF configuration based on the provided parameters. 
    Example:
    ```python
       cmf_init( type="local", 
                 path="/path/to/re",
                 git_remote_url="git@github.com:user/repo.git",
                 cmf_server_url="http://cmf-server"
                 neo4j_user", 
                 neo4j_password="password",
                 neo4j_uri="bolt://localhost:76"
               )
    ```
    Args: 
       type: Type of repository ("local", "minioS3", "amazonS3", "sshremote")
       path: Path for the local repository. 
       git_remote_url: Git remote URL for version control.
       cmf_server_url: CMF server URL.
       neo4j_user: Neo4j database username.
       neo4j_password: Neo4j database password.
       neo4j_uri: Neo4j database URI.
       url: URL for MinioS3 or AmazonS3.
       endpoint_url: Endpoint URL for MinioS3.
       access_key_id: Access key ID for MinioS3 or AmazonS3.
       secret_key: Secret key for MinioS3 or AmazonS3. 
       session_token: Session token for AmazonS3.
       user: SSH remote username.
       password: SSH remote password. 
       port: SSH remote port.
       osdf_path: OSDF Origin Path.
       osdf_cache: OSDF Cache Path (Optional).
       key_id: OSDF Key ID.
       key_path: OSDF Private Key Path.
       key_issuer: OSDF Key Issuer URL.
    Returns:
       Output based on the initialized repository type.
    """

    if type == "":
        return print("Error: Type is not provided")
    if type not in ["local","minioS3","amazonS3","sshremote","osdfremote"]:
        return print("Error: Type value is undefined"+ " "+type+".Expected: "+",".join(["local","minioS3","amazonS3","sshremote","osdfremote"]))

    if neo4j_user != "" and  neo4j_password != "" and neo4j_uri != "":
        pass
    elif neo4j_user == "" and  neo4j_password == "" and neo4j_uri == "":
        pass
    else:
        return print("Error: Enter all neo4j parameters.") 

    args={'path': path,
        'git_remote_url': git_remote_url,
        'url': url,
        'endpoint_url': endpoint_url,
        'access_key_id': access_key_id,
        'secret_key': secret_key,
        'session_token': session_token,
        'user': user,
        'password': password,
        'osdf_path': osdf_path,
        'osdf_cache': osdf_cache,
        'key_id': key_id,
        'key_path': key_path, 
        'key-issuer': key_issuer,
        }

    status_args=non_related_args(type, args)

    if type == "local" and path != "" and  git_remote_url != "" :
        """Initialize local repository"""
        output = _init_local(
            path, git_remote_url, cmf_server_url, neo4j_user, neo4j_password, neo4j_uri
        )
        if status_args != []:
            print("There are non-related arguments: "+",".join(status_args)+".Please remove them.")
        return output

    elif type == "minioS3" and url != "" and endpoint_url != "" and access_key_id != "" and secret_key != "" and git_remote_url != "":
        """Initialize minioS3 repository"""
        output = _init_minioS3(
            url,
            endpoint_url,
            access_key_id,
            secret_key,
            git_remote_url,
            cmf_server_url,
            neo4j_user,
            neo4j_password,
            neo4j_uri,
        )
        if status_args != []:
            print("There are non-related arguments: "+",".join(status_args)+".Please remove them.")
        return output

    elif type == "amazonS3" and url != "" and access_key_id != "" and secret_key != "" and git_remote_url != "":
        """Initialize amazonS3 repository"""
        output = _init_amazonS3(
            url,
            access_key_id,
            secret_key,
            session_token,
            git_remote_url,
            cmf_server_url,
            neo4j_user,
            neo4j_password,
            neo4j_uri,
        )
        if status_args != []:
            print("There are non-related arguments: "+",".join(status_args)+".Please remove them.")

        return output

    elif type == "sshremote" and path != "" and user != "" and port != 0 and password != "" and git_remote_url != "":
        """Initialize sshremote repository"""
        output = _init_sshremote(
            path,
            user,
            port,
            password,
            git_remote_url,
            cmf_server_url,
            neo4j_user,
            neo4j_password,
            neo4j_uri,
        )
        if status_args != []:
            print("There are non-related arguments: "+",".join(status_args)+".Please remove them.")

        return output

    elif type == "osdfremote" and osdf_path != "" and key_id != "" and key_path != "" and key_issuer != "" and git_remote_url != "":
        """Initialize osdfremote repository"""
        output = _init_osdfremote(
            osdf_path,
            osdf_cache,
            key_id,
            key_path,
            key_issuer,
            git_remote_url,
            cmf_server_url,
            neo4j_user,
            neo4j_password,
            neo4j_uri,
        )
        if status_args != []:
            print("There are non-related arguments: "+",".join(status_args)+".Please remove them.")

        return output

    else:
        print("Error: Enter all arguments")

metadata_push(pipeline_name, filepath='./mlmd', tensorboard_path='', execution_id='')

Pushes MLMD file to CMF-server. Example:

     result = metadata_push("example_pipeline", "mlmd_file", "3")

Args: pipeline_name: Name of the pipeline. filepath: Path to the MLMD file. execution_id: Optional execution ID. tensorboard_path: Path to tensorboard logs.

Returns:

Type	Description
	Response output from the _metadata_push function.

Source code in cmflib/cmf.py

def metadata_push(pipeline_name: str, filepath = "./mlmd", tensorboard_path: str = "", execution_id: str = ""):
    """ Pushes MLMD file to CMF-server.
    Example:
    ```python
         result = metadata_push("example_pipeline", "mlmd_file", "3")
    ```
    Args:
        pipeline_name: Name of the pipeline.
        filepath: Path to the MLMD file.
        execution_id: Optional execution ID.
        tensorboard_path: Path to tensorboard logs.

    Returns:
        Response output from the _metadata_push function.
    """
    # Required arguments:  pipeline_name
    # Optional arguments: Execution_ID, filepath (mlmd file path, tensorboard_path
    output = _metadata_push(pipeline_name, filepath, execution_id, tensorboard_path)
    return output

metadata_pull(pipeline_name, filepath='./mlmd', execution_id='')

Pulls MLMD file from CMF-server. Example:

     result = metadata_pull("example_pipeline", "./mlmd_directory", "execution_123") 

Args: pipeline_name: Name of the pipeline. filepath: File path to store the MLMD file. execution_id: Optional execution ID. Returns: Message from the _metadata_pull function.

Source code in cmflib/cmf.py

def metadata_pull(pipeline_name: str, filepath = "./mlmd", execution_id: str = ""):
    """ Pulls MLMD file from CMF-server. 
     Example: 
     ```python 
          result = metadata_pull("example_pipeline", "./mlmd_directory", "execution_123") 
     ``` 
     Args: 
        pipeline_name: Name of the pipeline. 
        filepath: File path to store the MLMD file. 
        execution_id: Optional execution ID. 
     Returns: 
        Message from the _metadata_pull function. 
     """
    # Required arguments:  pipeline_name 
    #Optional arguments: Execution_ID, filepath(file path to store mlmd file) 
    output = _metadata_pull(pipeline_name, filepath, execution_id)
    return output

artifact_pull(pipeline_name, filepath='./mlmd')

Pulls artifacts from the initialized repository.

Example:

     result = artifact_pull("example_pipeline", "./mlmd_directory")

Parameters:

Name	Type	Description	Default
pipeline_name	str	Name of the pipeline.	required
filepath		Path to store artifacts.	'./mlmd'

Returns: Output from the _artifact_pull function.

Source code in cmflib/cmf.py

def artifact_pull(pipeline_name: str, filepath = "./mlmd"):
    """ Pulls artifacts from the initialized repository.

    Example:
    ```python
         result = artifact_pull("example_pipeline", "./mlmd_directory")
    ```

    Args:
        pipeline_name: Name of the pipeline.
        filepath: Path to store artifacts.
    Returns:
        Output from the _artifact_pull function.
    """

    # Required arguments: Pipeline_name
    # Optional arguments: filepath( path to store artifacts)
    output = _artifact_pull(pipeline_name, filepath)
    return output

artifact_pull_single(pipeline_name, filepath, artifact_name)

Pulls a single artifact from the initialized repository. Example:

    result = artifact_pull_single("example_pipeline", "./mlmd_directory", "example_artifact") 

Args: pipeline_name: Name of the pipeline. filepath: Path to store the artifact. artifact_name: Name of the artifact. Returns: Output from the _artifact_pull_single function.

Source code in cmflib/cmf.py

def artifact_pull_single(pipeline_name: str, filepath: str, artifact_name: str):
    """ Pulls a single artifact from the initialized repository. 
    Example: 
    ```python 
        result = artifact_pull_single("example_pipeline", "./mlmd_directory", "example_artifact") 
    ```
    Args: 
       pipeline_name: Name of the pipeline. 
       filepath: Path to store the artifact. 
       artifact_name: Name of the artifact. 
    Returns:
       Output from the _artifact_pull_single function. 
    """

    # Required arguments: Pipeline_name
    # Optional arguments: filepath( path to store artifacts), artifact_name
    output = _artifact_pull_single(pipeline_name, filepath, artifact_name)
    return output

artifact_push(pipeline_name, filepath='./mlmd')

Pushes artifacts to the initialized repository.

Example:

     result = artifact_push("example_pipeline", "./mlmd_directory")

Args: pipeline_name: Name of the pipeline. filepath: Path to store the artifact. Returns: Output from the _artifact_push function.

Source code in cmflib/cmf.py

def artifact_push(pipeline_name: str, filepath = "./mlmd"):
    """ Pushes artifacts to the initialized repository.

    Example:
    ```python
         result = artifact_push("example_pipeline", "./mlmd_directory")
    ```
    Args: 
       pipeline_name: Name of the pipeline. 
       filepath: Path to store the artifact. 
    Returns:
        Output from the _artifact_push function.
    """

    output = _artifact_push(pipeline_name, filepath)
    return output