util module#

edgeimpulse.util.check_response_errors(request)[source]#

Check for standard errors and raise an exception with the details if found.

edgeimpulse.util.configure_generic_client(
key: str,
key_type: str = 'api',
host: str = 'https://studio.edgeimpulse.com/v1',
) ApiClient[source]#

Configure generic api client which the right key.

Parameters:
  • key (str) – api, jwt or jwt_http key.

  • key_type (str, optional) – Type of key. Defaults to api.

  • host (str) – API host url. Defaults to “https://studio.edgeimpulse.com/v1”.

Raises:

Exception – Unrecognized key_type

Returns:

Generic API client used in other generated APIs

Return type:

ApiClient

edgeimpulse.util.connect_websocket(token, host: str | None = None) Client[source]#

Connects to the websocket server.

Parameters:
  • token (str) – The authentication token.

  • host (str, optional) – The hostname. If None, API_ENDPOINT will be used.

Returns:

Websocket object.

Return type:

object

edgeimpulse.util.default_project_id_for(client: ApiClient) int[source]#

Derive project id from api_key used to configure generic client.

Parameters:

client (ApiClient) – Generic api client configured with a project api key. For jwt or jwt_http key use get_project_id_list().

Returns:

Project id

Return type:

int

edgeimpulse.util.encode_file_as_base64(filename: str)[source]#

Envode a file as base64.

edgeimpulse.util.get_organization_websocket(
client,
organization_id: int,
host: str | None = None,
) Client[source]#

Gets a websocket to listen to organization events.

Parameters:
  • client (object) – An API client object.

  • organization_id (int) – The ID of the organization.

  • host (str, optional) – The hostname. If None, API_ENDPOINT will be used.

Returns:

Websocket object.

Return type:

object

edgeimpulse.util.get_profile_devices(
client: ApiClient,
project_id: int | None = None,
) List[str][source]#

Pull a list of profile devices.

Parameters:
  • client (ApiClient) – Generic api client configured with a project api key

  • project_id – (int, Optional) Project id to get the targets for. Defaults to None.

Returns:

List of profile targets for project

Return type:

List[str]

edgeimpulse.util.get_project_deploy_targets(
client: ApiClient,
project_id: int | None = None,
) List[str][source]#

Pull a list of deploy targets.

Parameters:
  • client (ApiClient) – Generic api client configured with a project api key

  • project_id – (int, Optional) Project id to get the targets for. Defaults to None.

Returns:

List of deploy targets for project

Return type:

List[str]

edgeimpulse.util.get_project_websocket(
client,
project_id: int,
host: str | None = None,
) Client[source]#

Gets a websocket to listen to project events.

Parameters:
  • client (object) – An API client object.

  • project_id (int) – The ID of the project.

  • host (str, optional) – The hostname. If None, API_ENDPOINT will be used.

Returns:

Websocket object.

Return type:

object

edgeimpulse.util.get_user_agent(add_platform_info=False)[source]#

Get user agent string for API calls so we can track usage.

edgeimpulse.util.inspect_model(
model: Path | str | bytes | Any,
tempdir: str,
) Tuple[str, str][source]#

Load tflite model.

Parameters:
  • model – Supports a number of ways of representing a model including 1) A Path / str filename of an onnx file, a Keras saved_model.zip, a TensorFlow saved_model directory or saved tflite model. 2) a Keras model instance. 3) an ONNX model instance

  • tempdir – temp dir used to write saved form of any in memory model passed

Returns:

(model type suitable for API,

path to model saved suitable for API)

Return type:

Tuple(str)

edgeimpulse.util.inspect_representative_data(
data: Path | str | bytes | Any,
) str | None[source]#

Ensure representative data is saved to disk for upload.

Parameters:
  • data – either a str/Path to a numpy array or a np.ndarray

  • directory – a str with a directory path that the data should be saved to if necessary

Returns:

string path to the saved file

edgeimpulse.util.is_keras_model(model)[source]#

Check if model is a keras model.

edgeimpulse.util.is_numpy_array(array)[source]#

Check if array is a numpy array.

edgeimpulse.util.is_onnx_model(model)[source]#

Check if given model is an onnx model.

edgeimpulse.util.is_path_to_numpy_file(path)[source]#

Check if given path is a numpy file.

edgeimpulse.util.is_path_to_onnx_model(path)[source]#

Check if given path is a onnx file.

edgeimpulse.util.is_path_to_tf_saved_model_directory(model_dir)[source]#

Check if directory contains a saved model.

edgeimpulse.util.is_path_to_tf_saved_model_zipped(model)[source]#

Check if path is poiting to a zipped model.

edgeimpulse.util.is_type_accepted_by_open(path)[source]#

Check if given path is a. string or a Path.

edgeimpulse.util.make_zip_archive(saved_model_path)[source]#

Create zip archive from a model path.

edgeimpulse.util.numpy_installed() bool[source]#

Check if numpy is installed returns true or false.

edgeimpulse.util.onnx_installed() bool[source]#

Check if onnx is installed returns true or false.

edgeimpulse.util.pandas_installed() bool[source]#

Check if pandas is installed returns true or false.

edgeimpulse.util.poll(
jobs_client: JobsApi,
project_id: int,
job_id: int,
timeout_sec: float | None = None,
) GetJobResponse[source]#

Poll a specific job within a project until done or timmeout is reached.

Parameters:
  • jobs_client (JobsApi) – JobsApi client

  • project_id (int) – Project id number

  • job_id (int) – Job id to poll

  • timeout_sec (float, optional) – Optional timeout for polling.

Raises:
  • e – Unhandled exception from api

  • TimeoutException – Timeout waiting for result

Returns:

Structure containing job information

Return type:

GetJobResponse

edgeimpulse.util.run_job_until_completion(ws, job_id: int, data_cb=None, timeout_sec: int = 3600)[source]#

Runs a project or organization job until completion.

Parameters:
  • ws (object) – Websocket object.

  • job_id (int) – The ID of the job to run.

  • data_cb (callable, optional) – Callback function to handle job data.

  • timeout_sec (int, optional) – Number of seconds before timeing out the job with an exception. Default is 3600

Returns:

None

edgeimpulse.util.run_organization_job_until_completion(
organization_id: int,
job_id: int,
data_cb=None,
client=None,
timeout_sec: int = 3600,
) None[source]#

Runs an organization job until completion.

Parameters:
  • organization_id (int) – The ID of the organization.

  • job_id (int) – The ID of the job to run.

  • data_cb (callable, optional) – Callback function to handle job data.

  • client (object, optional) – An API client object. If None, a generic client will be configured.

  • timeout_sec (int, optional) – Number of seconds before timeing out the job with an exception. Default is 3600

Returns:

None

edgeimpulse.util.run_project_job_until_completion(
job_id: int,
data_cb=None,
client=None,
project_id: int | None = None,
timeout_sec: int = 3600,
) None[source]#

Runs a project job until completion.

Parameters:
  • job_id (int) – The ID of the job to run.

  • data_cb (callable, optional) – Callback function to handle job data.

  • client (object, optional) – An API client object. If None, a generic client will be configured.

  • project_id (int, optional) – The ID of the project. If not provided, a default project ID will be used.

  • timeout_sec (int, optional) – Number of seconds before timeing out the job with an exception. Default is 3600

Returns:

None

edgeimpulse.util.save_model(model: Path | str | bytes, directory: str) str[source]#

Save a machine learning model to the specified directory.

Parameters:
  • model (Union[Path, str, bytes]) – The machine learning model to be saved. It can be either a file path pointing to the location of the model file, a bytes object representing the model, or the model object itself.

  • directory (str) – The directory where the model will be saved.

Returns:

The path to the saved model or archive.

Return type:

str

Raises:
  • Exception – If the model is already located at the provided path.

  • InvalidModelException – If the model is of an unexpected type and cannot be processed.

Note

The function supports saving TensorFlow/Keras models in TensorFlow’s SavedModel format if TensorFlow is installed and the model is a Keras model. It also supports saving ONNX models if ONNX is installed and the model is in ONNX format.

edgeimpulse.util.save_representative_data(
data: Path | str | bytes,
directory: str,
) str[source]#

Save the representive data to a directory.

edgeimpulse.util.tensorflow_installed() bool[source]#

Check if tensorflow is installed returns true or false.

edgeimpulse.util.upload_pretrained_model_and_data(
tempdir: str,
client: ApiClient,
project_id: int,
model: Path | str | bytes | Any,
representative_data: Path | str | bytes | Any | None = None,
device: str | None = None,
timeout_sec: float | None = None,
) GetJobResponse[source]#

Upload a model and data to Edge Impulse servers.

Parameters:
  • tempdir (str) – Temporary directory to hold saved form of any model passed in

  • client (ApiClient) – Generic api client configured with a project api key. For jwt or jwt_http key use get_project_id_list().

  • project_id (int) – Project id number

  • model (Union[Path, str, bytes, Any]) – A machine learning model, or similarly represented computational graph. Can be Path or str denoting file path, Python bytes containing a model, or a Keras model instance.

  • representative_data (Optional[Union[Path, str, bytes, Any]], optional) – A numpy representative input dataset. Accepts either an in memory numpy array or the Path/str filename of a np.save .npy file.

  • device (Optional[str], optional) – An embedded processor for which to profile the model. A comprehensive list can be obtained via edgeimpulse.model.list_profile_devices().

  • timeout_sec (Optional[float], optional) – Optional timeout for polling.

Raises:
  • e – Unhandled exception from api

  • FileNotFoundError – File or directory not found

  • TimeoutException – Timeout waiting for result from server

Returns:

Structure containing job information

Return type:

GetJobResponse