Source code for

import logging
import pathlib
import typing
from typing import Union, Optional

from openeo.internal.documentation import openeo_process
from openeo.internal.graph_building import PGNode
from import _ProcessGraphAbstraction
from import BatchJob

if typing.TYPE_CHECKING:
    # Imports for type checking only (circular import issue at runtime).
    from openeo import Connection

_log = logging.getLogger(__name__)

[docs]class MlModel(_ProcessGraphAbstraction): """ A machine learning model. It is the result of a training procedure, e.g. output of a ``fit_...`` process, and can be used for prediction (classification or regression) with the corresponding ``predict_...`` process. .. versionadded:: 0.10.0 """ def __init__(self, graph: PGNode, connection: 'Connection'): super().__init__(pgnode=graph, connection=connection)
[docs] def save_ml_model(self, options: Optional[dict] = None): """ Saves a machine learning model as part of a batch job. :param options: Additional parameters to create the file(s). """ pgnode = PGNode( process_id="save_ml_model", arguments={"data": self, "options": options or {}} ) return MlModel(graph=pgnode, connection=self._connection)
[docs] @staticmethod @openeo_process def load_ml_model(connection: "Connection", id: Union[str, BatchJob]) -> "MlModel": """ Loads a machine learning model from a STAC Item. :param connection: connection object :param id: STAC item reference, as URL, batch job (id) or user-uploaded file :return: .. versionadded:: 0.10.0 """ if isinstance(id, BatchJob): id = id.job_id return MlModel(graph=PGNode(process_id="load_ml_model", id=id), connection=connection)
[docs] def execute_batch( self, outputfile: Union[str, pathlib.Path], print=print, max_poll_interval=60, connection_retry_interval=30, job_options=None, ) -> BatchJob: """ Evaluate the process graph by creating a batch job, and retrieving the results when it is finished. This method is mostly recommended if the batch job is expected to run in a reasonable amount of time. For very long running jobs, you probably do not want to keep the client running. :param job_options: :param outputfile: The path of a file to which a result can be written :param out_format: (optional) Format of the job result. :param format_options: String Parameters for the job result format """ job = self.create_job(additional=job_options) return job.run_synchronous( # TODO #135 support multi file result sets too outputfile=outputfile, print=print, max_poll_interval=max_poll_interval, connection_retry_interval=connection_retry_interval )
[docs] def create_job(self, **kwargs) -> BatchJob: """ Sends a job to the backend and returns a ClientJob instance. See :py:meth:`Connection.create_job` for additional arguments (e.g. to set job title, description, ...) :return: resulting job. """ pg = self if pg.result_node().process_id not in {"save_ml_model"}: _log.warning("Process graph has no final `save_ml_model`. Adding it automatically.") pg = pg.save_ml_model() return self._connection.create_job(process_graph=pg.flat_graph(), **kwargs)