Ocean Compute Tech Details

Technical details about OceanCompute functions

Using this class, we are able to manipulate a compute job, run it on Ocean environment and retrieve the results after the execution is finished.

Start Compute Job

start(self, consumer_wallet, dataset: ComputeInput, compute_environment: str, algorithm: Optional[ComputeInput] = None, algorithm_meta: Optional[AlgorithmMetadata] = None, algorithm_algocustomdata: Optional[dict] = None, additional_datasets: List[ComputeInput] = []) -> str

Starts a compute job.

It can be called within Ocean Compute class.

Parameters

consumer_wallet - the eth Account of consumer who pays & starts for compute job.
dataset - ComputeInput object, each of them includes mandatory the DDO and service.
compute_environment - string that represents the ID from the chosen C2D environment.
additional_datasets - list of ComputeInput objects for additional datasets in case of starting a compute job for multiple datasets.

Optional parameters

algorithm - ComputeInput object, each of them includes mandatory the DDO and service for algorithm.
algorithm_meta - either provide just the algorithm metadata as AlgorithmMetadata.
algorithm_algocustomedata - additional user data for the algorithm as dictionary.

Returns

str

Returns a string type job ID.

Defined in

Source code

 @enforce_types
    def start(
        self,
        consumer_wallet,
        dataset: ComputeInput,
        compute_environment: str,
        algorithm: Optional[ComputeInput] = None,
        algorithm_meta: Optional[AlgorithmMetadata] = None,
        algorithm_algocustomdata: Optional[dict] = None,
        additional_datasets: List[ComputeInput] = [],
    ) -> str:
        metadata_cache_uri = self._config_dict.get("METADATA_CACHE_URI")
        ddo = Aquarius.get_instance(metadata_cache_uri).get_ddo(dataset.did)
        service = ddo.get_service_by_id(dataset.service_id)
        assert (
            ServiceTypes.CLOUD_COMPUTE == service.type
        ), "service at serviceId is not of type compute service."

        consumable_result = is_consumable(
            ddo,
            service,
            {"type": "address", "value": consumer_wallet.address},
            with_connectivity_check=True,
        )
        if consumable_result != ConsumableCodes.OK:
            raise AssetNotConsumable(consumable_result)

        # Start compute job
        job_info = self._data_provider.start_compute_job(
            dataset_compute_service=service,
            consumer=consumer_wallet,
            dataset=dataset,
            compute_environment=compute_environment,
            algorithm=algorithm,
            algorithm_meta=algorithm_meta,
            algorithm_custom_data=algorithm_algocustomdata,
            input_datasets=additional_datasets,
        )
        return job_info["jobId"]

Compute Job Status

status(self, ddo: DDO, service: Service, job_id: str, wallet) -> Dict[str, Any]

Gets status of the compute job.

It can be called within Ocean Compute class.

Parameters

ddo - DDO offering the compute service of this job
service - Service object of compute
job_id - ID of the compute job
wallet - eth Account which initiated the compute job

Returns

Dict[str, Any]

A dictionary which contains the status for an existing compute job, keys are (ok, status, statusText).

Defined in

Source code

@enforce_types
    def status(self, ddo: DDO, service: Service, job_id: str, wallet) -> Dict[str, Any]:
        """
        Gets job status.

        :param ddo: DDO offering the compute service of this job
        :param service: compute service of this job
        :param job_id: str id of the compute job
        :param wallet: Wallet instance
        :return: dict the status for an existing compute job, keys are (ok, status, statusText)
        """
        job_info = self._data_provider.compute_job_status(
            ddo.did, job_id, service, wallet
        )
        job_info.update({"ok": job_info.get("status") not in (31, 32, None)})

        return job_info

Compute Job Result

result(self, ddo: DDO, service: Service, job_id: str, index: int, wallet ) -> Dict[str, Any]

Gets compute job result.

It can be called within Ocean Compute class.

Parameters

ddo - DDO offering the compute service of this job
service - Service object of compute
job_id - ID of the compute job
index - compute result index
wallet - eth Account which initiated the compute job

Returns

Dict[str, Any]

A dictionary wich contains the results/logs urls for an existing compute job, keys are (did, urls, logs).

Defined in

Source code

@enforce_types
    def result(
        self, ddo: DDO, service: Service, job_id: str, index: int, wallet
    ) -> Dict[str, Any]:
        """
        Gets job result.

        :param ddo: DDO offering the compute service of this job
        :param service: compute service of this job
        :param job_id: str id of the compute job
        :param index: compute result index
        :param wallet: Wallet instance
        :return: dict the results/logs urls for an existing compute job, keys are (did, urls, logs)
        """
        result = self._data_provider.compute_job_result(job_id, index, service, wallet)

        return result

Compute Job Result Logs

compute_job_result_logs(self, ddo: DDO, service: Service, job_id: str, wallet, log_type="output") -> Dict[str, Any]

Gets job output if exists.

It can be called within Ocean Compute class.

Parameters

ddo - DDO offering the compute service of this job
service - Service object of compute
job_id - ID of the compute job
wallet - eth Account which initiated the compute job
log_type - string which selects what kind of logs to display. Default "output"

Returns

Dict[str, Any]

A dictionary which includes the results/logs urls for an existing compute job, keys are (did, urls, logs).

Defined in

Source code

@enforce_types
    def compute_job_result_logs(
        self,
        ddo: DDO,
        service: Service,
        job_id: str,
        wallet,
        log_type="output",
    ) -> Dict[str, Any]:
        """
        Gets job output if exists.

        :param ddo: DDO offering the compute service of this job
        :param service: compute service of this job
        :param job_id: str id of the compute job
        :param wallet: Wallet instance
        :return: dict the results/logs urls for an existing compute job, keys are (did, urls, logs)
        """
        result = self._data_provider.compute_job_result_logs(
            ddo, job_id, service, wallet, log_type
        )

        return result

Stop Compute Job

stop(self, ddo: DDO, service: Service, job_id: str, wallet) -> Dict[str, Any]

Attempts to stop the running compute job.

It can be called within Ocean Compute class.

Parameters

ddo - DDO offering the compute service of this job
service - Service object of compute
job_id - ID of the compute job
wallet - eth Account which initiated the compute job

Returns

Dict[str, Any]

A dictionary which contains the status for the stopped compute job, keys are (ok, status, statusText).

Defined in

Source code

@enforce_types
    def stop(self, ddo: DDO, service: Service, job_id: str, wallet) -> Dict[str, Any]:
        """
        Attempt to stop the running compute job.

        :param ddo: DDO offering the compute service of this job
        :param job_id: str id of the compute job
        :param wallet: Wallet instance
        :return: dict the status for the stopped compute job, keys are (ok, status, statusText)
        """
        job_info = self._data_provider.stop_compute_job(
            ddo.did, job_id, service, wallet
        )
        job_info.update({"ok": job_info.get("status") not in (31, 32, None)})
        return job_info

Get Priced C2D Environments

get_c2d_environments(self, service_endpoint: str, chain_id: int)

Get list of compute environments.

It can be called within Ocean Compute class.

Parameters

service_endpoint - string Provider URL that is stored in compute service.
chain_id - using Provider multichain, chain_id is required to specify the network for your environment. It has int type.

Returns

list

A list of objects containing information about each compute environment. For each compute environment, these are the following keys: (id, feeToken, priceMin, consumerAddress, lastSeen, namespace, status).

Defined in

Source code

    @enforce_types
    def get_c2d_environments(self, service_endpoint: str, chain_id: int):
        return DataServiceProvider.get_c2d_environments(service_endpoint, chain_id)

Get Free C2D Environments

get_free_c2d_environment(self, service_endpoint: str, chain_id)

Get list of free compute environments.

Important thing is that not all Providers contain free environments (priceMin = 0).

It can be called within Ocean Compute class.

Parameters

service_endpoint - string Provider URL that is stored in compute service.
chain_id - using Provider multichain, chain_id is required to specify the network for your environment. It has int type.

Returns

list

Defined in

Source code

@enforce_types
    def get_free_c2d_environment(self, service_endpoint: str, chain_id):
        environments = self.get_c2d_environments(service_endpoint, chain_id)
        return next(env for env in environments if float(env["priceMin"]) == float(0))

Last updated 1 year ago

Was this helpful?

@enforce_types def start( self, consumer_wallet, dataset: ComputeInput, compute_environment: str, algorithm: Optional[ComputeInput] = None, algorithm_meta: Optional[AlgorithmMetadata] = None, algorithm_algocustomdata: Optional[dict] = None, additional_datasets: List[ComputeInput] = [], ) -> str: metadata_cache_uri = self._config_dict.get("METADATA_CACHE_URI") ddo = Aquarius.get_instance(metadata_cache_uri).get_ddo(dataset.did) service = ddo.get_service_by_id(dataset.service_id) assert ( ServiceTypes.CLOUD_COMPUTE == service.type ), "service at serviceId is not of type compute service." consumable_result = is_consumable( ddo, service, {"type": "address", "value": consumer_wallet.address}, with_connectivity_check=True, ) if consumable_result != ConsumableCodes.OK: raise AssetNotConsumable(consumable_result) # Start compute job job_info = self._data_provider.start_compute_job( dataset_compute_service=service, consumer=consumer_wallet, dataset=dataset, compute_environment=compute_environment, algorithm=algorithm, algorithm_meta=algorithm_meta, algorithm_custom_data=algorithm_algocustomdata, input_datasets=additional_datasets, ) return job_info["jobId"]

@enforce_types def status(self, ddo: DDO, service: Service, job_id: str, wallet) -> Dict[str, Any]: """ Gets job status. :param ddo: DDO offering the compute service of this job :param service: compute service of this job :param job_id: str id of the compute job :param wallet: Wallet instance :return: dict the status for an existing compute job, keys are (ok, status, statusText) """ job_info = self._data_provider.compute_job_status( ddo.did, job_id, service, wallet ) job_info.update({"ok": job_info.get("status") not in (31, 32, None)}) return job_info

@enforce_types def result( self, ddo: DDO, service: Service, job_id: str, index: int, wallet ) -> Dict[str, Any]: """ Gets job result. :param ddo: DDO offering the compute service of this job :param service: compute service of this job :param job_id: str id of the compute job :param index: compute result index :param wallet: Wallet instance :return: dict the results/logs urls for an existing compute job, keys are (did, urls, logs) """ result = self._data_provider.compute_job_result(job_id, index, service, wallet) return result

@enforce_types def compute_job_result_logs( self, ddo: DDO, service: Service, job_id: str, wallet, log_type="output", ) -> Dict[str, Any]: """ Gets job output if exists. :param ddo: DDO offering the compute service of this job :param service: compute service of this job :param job_id: str id of the compute job :param wallet: Wallet instance :return: dict the results/logs urls for an existing compute job, keys are (did, urls, logs) """ result = self._data_provider.compute_job_result_logs( ddo, job_id, service, wallet, log_type ) return result

@enforce_types def stop(self, ddo: DDO, service: Service, job_id: str, wallet) -> Dict[str, Any]: """ Attempt to stop the running compute job. :param ddo: DDO offering the compute service of this job :param job_id: str id of the compute job :param wallet: Wallet instance :return: dict the status for the stopped compute job, keys are (ok, status, statusText) """ job_info = self._data_provider.stop_compute_job( ddo.did, job_id, service, wallet ) job_info.update({"ok": job_info.get("status") not in (31, 32, None)}) return job_info

@enforce_types def get_free_c2d_environment(self, service_endpoint: str, chain_id): environments = self.get_c2d_environments(service_endpoint, chain_id) return next(env for env in environments if float(env["priceMin"]) == float(0))