util module

check_response_errors

edgeimpulse.util.check_response_errors(
		request
)

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

Parameters

• request

configure_generic_client

edgeimpulse.util.configure_generic_client(
		key: str,
		key_type: str = 'api',
		host: str = 'https://studio.edgeimpulse.com/v1'
) ‑> edgeimpulse_api.api_client.ApiClient

Configure generic api client which the right key.

Parameters

• key: str

• key_type: str = 'api'

• host: str = 'https://studio.edgeimpulse.com/v1'

Return

edgeimpulse_api.api_client.ApiClient

connect_websocket

edgeimpulse.util.connect_websocket(
		token,
		host: str = None
) ‑> socketio.client.Client

Connects to the websocket server.

Parameters: token (str): The authentication token. host (str, optional): The hostname. If None, API_ENDPOINT will be used.

Returns: object: Websocket object.

Parameters

• token

• host: str = None

Return

socketio.client.Client

default_project_id_for

edgeimpulse.util.default_project_id_for(
		client: edgeimpulse_api.api_client.ApiClient
) ‑> int

Derive project id from api_key used to configure generic client.

Parameters

• client: edgeimpulse_api.api_client.ApiClient

Return

int

encode_file_as_base64

edgeimpulse.util.encode_file_as_base64(
		filename: str
)

Envode a file as base64.

Parameters

• filename: str

get_organization_websocket

edgeimpulse.util.get_organization_websocket(
		client,
		organization_id: int,
		host: str = None
) ‑> socketio.client.Client

Gets a websocket to listen to organization events.

Parameters

• client

• organization_id: int

• host: str = None

Return

socketio.client.Client

get_profile_devices

edgeimpulse.util.get_profile_devices(
		client: edgeimpulse_api.api_client.ApiClient,
		project_id: Optional[int] = None
) ‑> List[str]

Pull a list of profile devices.

Parameters

• client: edgeimpulse_api.api_client.ApiClient

• project_id: Optional[int] = None

Return

List[str]

get_project_deploy_targets

edgeimpulse.util.get_project_deploy_targets(
		client: edgeimpulse_api.api_client.ApiClient,
		project_id: Optional[int] = None
) ‑> List[str]

Pull a list of deploy targets.

Parameters

• client: edgeimpulse_api.api_client.ApiClient

• project_id: Optional[int] = None

Return

List[str]

get_project_websocket

edgeimpulse.util.get_project_websocket(
		client,
		project_id: int,
		host: str = None
) ‑> socketio.client.Client

Gets a websocket to listen to project events.

Parameters

• client

• project_id: int

• host: str = None

Return

socketio.client.Client

get_user_agent

edgeimpulse.util.get_user_agent(
		add_platform_info=False
)

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

Parameters

• add_platform_info=False

inspect_model

edgeimpulse.util.inspect_model(
		model: Union[pathlib.Path, str, bytes, Any],
		tempdir: str
) ‑> Tuple[str, str]

Load tflite model.

Parameters

• model: Union[pathlib.Path, str, bytes, Any]

• tempdir: str

Return

Tuple[str, str]

inspect_representative_data

edgeimpulse.util.inspect_representative_data(
		data: Union[pathlib.Path, str, bytes, Any]
) ‑> Optional[str]

Ensure representative data is saved to disk for upload.

Parameters

• data: Union[pathlib.Path, str, bytes, Any]

Return

Optional[str]

is_keras_model

edgeimpulse.util.is_keras_model(
		model
)

Check if model is a keras model.

Parameters

• model

is_numpy_array

edgeimpulse.util.is_numpy_array(
		array
)

Check if array is a numpy array.

Parameters

• array

is_onnx_model

edgeimpulse.util.is_onnx_model(
		model
)

Check if given model is an onnx model.

Parameters

• model

is_path_to_numpy_file

edgeimpulse.util.is_path_to_numpy_file(
		path
)

Check if given path is a numpy file.

Parameters

• path

is_path_to_onnx_model

edgeimpulse.util.is_path_to_onnx_model(
		path
)

Check if given path is a onnx file.

Parameters

• path

is_path_to_tf_saved_model_directory

edgeimpulse.util.is_path_to_tf_saved_model_directory(
		model_dir
)

Check if directory contains a saved model.

Parameters

• model_dir

is_path_to_tf_saved_model_zipped

edgeimpulse.util.is_path_to_tf_saved_model_zipped(
		model
)

Check if path is poiting to a zipped model.

Parameters

• model

is_type_accepted_by_open

edgeimpulse.util.is_type_accepted_by_open(
		path
)

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

Parameters

• path

make_zip_archive

edgeimpulse.util.make_zip_archive(
		saved_model_path
)

Create zip archive from a model path.

Parameters

• saved_model_path

numpy_installed

edgeimpulse.util.numpy_installed(
		
) ‑> bool

Check if numpy is installed returns true or false.

Return

bool

onnx_installed

edgeimpulse.util.onnx_installed(
		
) ‑> bool

Check if onnx is installed returns true or false.

Return

bool

pandas_installed

edgeimpulse.util.pandas_installed(
		
) ‑> bool

Check if pandas is installed returns true or false.

Return

bool

poll

edgeimpulse.util.poll(
		jobs_client: edgeimpulse_api.api.jobs_api.JobsApi,
		project_id: int,
		job_id: int,
		timeout_sec: Optional[float] = None
) ‑> edgeimpulse_api.models.get_job_response.GetJobResponse

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

Parameters

• jobs_client: edgeimpulse_api.api.jobs_api.JobsApi

• project_id: int

• job_id: int

• timeout_sec: Optional[float] = None

Return

edgeimpulse_api.models.get_job_response.GetJobResponse

run_job_until_completion

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

Runs a project or organization job until completion.

Parameters

• ws

• job_id: int

• data_cb=None

• timeout_sec: int = 3600

run_organization_job_until_completion

edgeimpulse.util.run_organization_job_until_completion(
		organization_id: int,
		job_id: int,
		data_cb=None,
		client=None,
		timeout_sec: int = 3600
) ‑> None

Runs an organization job until completion.

Parameters

• organization_id: int

• job_id: int

• data_cb=None

• client=None

• timeout_sec: int = 3600

Return

None

run_project_job_until_completion

edgeimpulse.util.run_project_job_until_completion(
		job_id: int,
		data_cb=None,
		client=None,
		project_id: int = None,
		timeout_sec: int = 3600
) ‑> None

Runs a project job until completion.

Parameters

• job_id: int

• data_cb=None

• client=None

• project_id: int = None

• timeout_sec: int = 3600

Return

None

save_model

edgeimpulse.util.save_model(
		model: Union[pathlib.Path, str, bytes],
		directory: str
) ‑> str

Save a machine learning model to the specified directory.

Parameters

• model: Union[pathlib.Path, str, bytes]

• directory: str

Return

str

save_representative_data

edgeimpulse.util.save_representative_data(
		data: Union[pathlib.Path, str, bytes],
		directory: str
) ‑> str

Save the representive data to a directory.

Parameters

• data: Union[pathlib.Path, str, bytes]

• directory: str

Return

str

tensorflow_installed

edgeimpulse.util.tensorflow_installed(
		
) ‑> bool

Check if tensorflow is installed returns true or false.

Return

bool

upload_pretrained_model_and_data

edgeimpulse.util.upload_pretrained_model_and_data(
		tempdir: str,
		client: edgeimpulse_api.api_client.ApiClient,
		project_id: int,
		model: Union[pathlib.Path, str, bytes, Any],
		representative_data: Union[pathlib.Path, str, bytes, Any, None] = None,
		device: Optional[str] = None,
		timeout_sec: Optional[float] = None
) ‑> edgeimpulse_api.models.get_job_response.GetJobResponse

Upload a model and data to Edge Impulse servers.

Parameters

• tempdir: str

• client: edgeimpulse_api.api_client.ApiClient

• project_id: int

• model: Union[pathlib.Path, str, bytes, Any]

• representative_data: Union[pathlib.Path, str, bytes, Any, None] = None

• device: Optional[str] = None

• timeout_sec: Optional[float] = None

Return

edgeimpulse_api.models.get_job_response.GetJobResponse