python/replicate/experiment.py
def checkpoint(
        self,
        path: Optional[str] = None,
        step: Optional[int] = None,
        metrics: Optional[Dict[str, Any]] = None,
        primary_metric: Optional[Tuple[str, str]] = None,
        quiet: bool = False,
    ) -> Optional[Checkpoint]:
        """
        Create a checkpoint within this experiment.

        This saves the metrics at this point, and makes a copy of the file or directory passed to `path`, which could be weights or any other artifact.
        """
        # TODO(bfirsh): display warning if primary_metric changes in an experiment
        # FIXME: store as tuple throughout for consistency?
        primary_metric_dict: Optional[PrimaryMetric] = None
        if primary_metric is not None:
            if len(primary_metric) != 2:
                # Not having a primary_metric isn't a fatal problem
                console.error(
                    "Not setting primary_metric on checkpoint: it must be a tuple of (name, goal), where name corresponds to a metric key, and goal is either 'maximize' or 'minimize'"
                )
            else:
                primary_metric_dict = {
                    "name": primary_metric[0],
                    "goal": primary_metric[1],
                }

        # Auto-increment step if not provided
        if step is None:
            step = self._step + 1
        # Remember the current step
        self._step = step

        checkpoint = Checkpoint(
            id=random_hash(),
            created=datetime.datetime.utcnow(),
            path=path,
            step=step,
            metrics=metrics,
            primary_metric=primary_metric_dict,
        )
        if not quiet:
            if path is None:
                console.info("Creating checkpoint {}".format(checkpoint.short_id()))
            else:
                console.info(
                    "Creating checkpoint {}: copying '{}' to '{}'...".format(
                        checkpoint.short_id(),
                        checkpoint.path,
                        self._project._get_repository().root_url(),
                    )
                )

        errors = checkpoint.validate()
        if errors:
            for error in errors:
                console.error("Not saving checkpoint: " + error)
            return checkpoint

        checkpoint._experiment = self

        # Upload files before writing metadata so if it is cancelled, there isn't metadata pointing at non-existent data
        if checkpoint.path is not None:
            tar_path = checkpoint._repository_tar_path()
            repository = self._project._get_repository()
            repository.put_path_tar(self._project.directory, tar_path, checkpoint.path)

        self.checkpoints.append(checkpoint)
        self.save()

        if self._heartbeat is not None:
            self._heartbeat.ensure_running()

        return checkpoint

        Create a checkpoint within this experiment.

        This saves the metrics at this point, and makes a copy of the file or directory passed to `path`, which could be weights or any other artifact.
        
Create a checkpoint within this experiment.
This saves the metrics at this point, and makes a copy of the file or directory passed to `path`, which could be weights or any other artifact.
Create a checkpoint within this experiment . This saves the metrics at this point and makes a copy of the file or directory passed to ` path ` which could be weights or any other artifact .
def checkpoint(
        self,
        path: Optional[str] = None,
        step: Optional[int] = None,
        metrics: Optional[Dict[str, Any]] = None,
        primary_metric: Optional[Tuple[str, str]] = None,
        quiet: bool = False,
    ) -> Optional[Checkpoint]:
        primary_metric_dict: Optional[PrimaryMetric] = None
        if primary_metric is not None:
            if len(primary_metric) != 2:
                console.error(
                    "Not setting primary_metric on checkpoint: it must be a tuple of (name, goal), where name corresponds to a metric key, and goal is either 'maximize' or 'minimize'"
                )
            else:
                primary_metric_dict = {
                    "name": primary_metric[0],
                    "goal": primary_metric[1],
                }
        if step is None:
            step = self._step + 1
        self._step = step
        checkpoint = Checkpoint(
            id=random_hash(),
            created=datetime.datetime.utcnow(),
            path=path,
            step=step,
            metrics=metrics,
            primary_metric=primary_metric_dict,
        )
        if not quiet:
            if path is None:
                console.info("Creating checkpoint {}".format(checkpoint.short_id()))
            else:
                console.info(
                    "Creating checkpoint {}: copying '{}' to '{}'...".format(
                        checkpoint.short_id(),
                        checkpoint.path,
                        self._project._get_repository().root_url(),
                    )
                )
        errors = checkpoint.validate()
        if errors:
            for error in errors:
                console.error("Not saving checkpoint: " + error)
            return checkpoint
        checkpoint._experiment = self
        if checkpoint.path is not None:
            tar_path = checkpoint._repository_tar_path()
            repository = self._project._get_repository()
            repository.put_path_tar(self._project.directory, tar_path, checkpoint.path)
        self.checkpoints.append(checkpoint)
        self.save()
        if self._heartbeat is not None:
            self._heartbeat.ensure_running()
        return checkpoint
def checkpoint ( self , path : Optional [ str ] = None , step : Optional [ int ] = None , metrics : Optional [ Dict [ str , Any ] ] = None , primary_metric : Optional [ Tuple [ str , str ] ] = None , quiet : bool = False , ) -> Optional [ Checkpoint ] : primary_metric_dict : Optional [ PrimaryMetric ] = None if primary_metric is not None : if len ( primary_metric ) != 2 : console . error ( "Not setting primary_metric on checkpoint: it must be a tuple of (name, goal), where name corresponds to a metric key, and goal is either 'maximize' or 'minimize'" ) else : primary_metric_dict = { "name" : primary_metric [ 0 ] , "goal" : primary_metric [ 1 ] , } if step is None : step = self . _step + 1 self . _step = step checkpoint = Checkpoint ( id = random_hash ( ) , created = datetime . datetime . utcnow ( ) , path = path , step = step , metrics = metrics , primary_metric = primary_metric_dict , ) if not quiet : if path is None : console . info ( "Creating checkpoint {}" . format ( checkpoint . short_id ( ) ) ) else : console . info ( "Creating checkpoint {}: copying '{}' to '{}'..." . format ( checkpoint . short_id ( ) , checkpoint . path , self . _project . _get_repository ( ) . root_url ( ) , ) ) errors = checkpoint . validate ( ) if errors : for error in errors : console . error ( "Not saving checkpoint: " + error ) return checkpoint checkpoint . _experiment = self if checkpoint . path is not None : tar_path = checkpoint . _repository_tar_path ( ) repository = self . _project . _get_repository ( ) repository . put_path_tar ( self . _project . directory , tar_path , checkpoint . path ) self . checkpoints . append ( checkpoint ) self . save ( ) if self . _heartbeat is not None : self . _heartbeat . ensure_running ( ) return checkpoint
Create a checkpoint within this experiment.
Create a checkpoint within this experiment .
"""
        Create a checkpoint within this experiment.

        This saves the metrics at this point, and makes a copy of the file or directory passed to `path`, which could be weights or any other artifact.
        """ # TODO(bfirsh): display warning if primary_metric changes in an experiment # FIXME: store as tuple throughout for consistency? # Not having a primary_metric isn't a fatal problem # Auto-increment step if not provided # Remember the current step # Upload files before writing metadata so if it is cancelled, there isn't metadata pointing at non-existent data
self
path
Optional[str]
step
Optional[int]
metrics
Optional[Dict[str, Any]]
primary_metric
Optional[Tuple[str, str]]
quiet
bool