inference_sdk API Reference¶

class InferenceSDKDeprecationWarning(Warning):
    """Class used for warning of deprecated features in the Inference SDK"""

    pass

class RemoteProcessingTimeCollector:
    """Thread-safe collector for GPU processing times from remote execution responses.

    A single instance is shared across all threads handling a single request.
    Each entry stores a model_id alongside the processing time.

    Uses threading.Lock (not asyncio.Lock) because add() is only called from
    synchronous worker threads (ThreadPoolExecutor). The middleware reads via
    drain() after await call_next() returns, at which point all worker threads
    have completed — so there is no contention in the async context.
    """

    def __init__(self):
        self._entries: list = []  # list of (model_id, time) tuples
        self._lock = threading.Lock()

    def add(self, processing_time: float, model_id: str = "unknown") -> None:
        with self._lock:
            self._entries.append((model_id, processing_time))

    def drain(self) -> list:
        """Atomically return all entries and clear the internal list."""
        with self._lock:
            entries = self._entries
            self._entries = []
            return entries

    def has_data(self) -> bool:
        with self._lock:
            return len(self._entries) > 0

    def summarize(self, max_detail_bytes: int = 4096) -> Tuple[float, Optional[str]]:
        """Atomically drain entries and return (total_time, entries_json_or_none).

        Returns the total processing time and a JSON string of individual entries.
        If the JSON exceeds max_detail_bytes, the detail string is omitted (None).
        """
        entries = self.drain()
        total = sum(t for _, t in entries)
        detail = json.dumps([{"m": m, "t": t} for m, t in entries])
        if len(detail) > max_detail_bytes:
            detail = None
        return total, detail

class InferenceHTTPClient:
    """HTTP client for making inference requests to Roboflow's API.

    This client handles authentication, request formatting, and error handling for
    interacting with Roboflow's inference endpoints. It supports both synchronous
    and asynchronous requests.

    Attributes:
        inference_configuration (InferenceConfiguration): Configuration settings for
            inference requests.
        client_mode (HTTPClientMode): The API version mode being used (V0 or V1).
        selected_model (Optional[str]): Currently selected model identifier, if any.

    Example:
        ```python
        from inference_sdk import InferenceHTTPClient

        client = InferenceHTTPClient(
            api_url="http://localhost:9001", # use local inference server
            # api_key="<YOUR API KEY>" # optional to access your private data and models
        )

        result = client.run_workflow(
            workspace_name="roboflow-docs",
            workflow_id="model-comparison",
            images={
                "image": "https://media.roboflow.com/workflows/examples/bleachers.jpg"
            },
            parameters={
                "model1": "yolov8n-640",
                "model2": "yolov11n-640"
            }
        )
        ```
    """

    @classmethod
    def init(
        cls,
        api_url: str,
        api_key: Optional[str] = None,
    ) -> "InferenceHTTPClient":
        """Initialize a new InferenceHTTPClient instance.

        Args:
            api_url (str): The base URL for the inference API.
            api_key (Optional[str], optional): API key for authentication. Defaults to None.

        Returns:
            InferenceHTTPClient: A new instance of the InferenceHTTPClient.
        """
        return cls(api_url=api_url, api_key=api_key)

    def __init__(
        self,
        api_url: str,
        api_key: Optional[str] = None,
    ):
        """Initialize a new InferenceHTTPClient instance.

        Args:
            api_url (str): The base URL for the inference API.
            api_key (Optional[str], optional): API key for authentication. Defaults to None.
        """
        self.__api_url = api_url
        self.__api_key = api_key
        self.__inference_configuration = InferenceConfiguration.init_default()
        self.__client_mode = _determine_client_mode(api_url=api_url)
        self.__selected_model: Optional[str] = None
        self.__webrtc_client: Optional["WebRTCClient"] = None

    @property
    def inference_configuration(self) -> InferenceConfiguration:
        """Get the current inference configuration.

        Returns:
            InferenceConfiguration: The current inference configuration settings.
        """
        return self.__inference_configuration

    @property
    def client_mode(self) -> HTTPClientMode:
        """Get the current client mode.

        Returns:
            HTTPClientMode: The current API version mode (V0 or V1).
        """
        return self.__client_mode

    @property
    def selected_model(self) -> Optional[str]:
        """Get the currently selected model identifier.

        Returns:
            Optional[str]: The identifier of the currently selected model, if any.
        """
        return self.__selected_model

    @property
    def webrtc(self) -> "WebRTCClient":
        """Lazy accessor for the WebRTC client namespace.

        Returns:
            WebRTCClient: Namespaced WebRTC API bound to this HTTP client.
        """
        from inference_sdk.webrtc.client import WebRTCClient

        if self.__webrtc_client is None:
            self.__webrtc_client = WebRTCClient(self.__api_url, self.__api_key)
        return self.__webrtc_client

    @contextmanager
    def use_configuration(
        self, inference_configuration: InferenceConfiguration
    ) -> Generator["InferenceHTTPClient", None, None]:
        """Temporarily use a different inference configuration.

        Args:
            inference_configuration (InferenceConfiguration): The temporary configuration to use.

        Yields:
            Generator[InferenceHTTPClient, None, None]: The client instance with temporary configuration.
        """
        previous_configuration = self.__inference_configuration
        self.__inference_configuration = inference_configuration
        try:
            yield self
        finally:
            self.__inference_configuration = previous_configuration

    def configure(
        self, inference_configuration: InferenceConfiguration
    ) -> "InferenceHTTPClient":
        """Configure the client with new inference settings.

        Args:
            inference_configuration (InferenceConfiguration): The new configuration to apply.

        Returns:
            InferenceHTTPClient: The client instance with updated configuration.
        """
        self.__inference_configuration = inference_configuration
        return self

    def select_api_v0(self) -> "InferenceHTTPClient":
        """Select API version 0 for client operations.

        Returns:
            InferenceHTTPClient: The client instance with API v0 selected.
        """
        self.__client_mode = HTTPClientMode.V0
        return self

    def select_api_v1(self) -> "InferenceHTTPClient":
        """Select API version 1 for client operations.

        Returns:
            InferenceHTTPClient: The client instance with API v1 selected.
        """
        self.__client_mode = HTTPClientMode.V1
        return self

    @contextmanager
    def use_api_v0(self) -> Generator["InferenceHTTPClient", None, None]:
        """Temporarily use API version 0 for client operations.

        Yields:
            Generator[InferenceHTTPClient, None, None]: The client instance temporarily using API v0.
        """
        previous_client_mode = self.__client_mode
        self.__client_mode = HTTPClientMode.V0
        try:
            yield self
        finally:
            self.__client_mode = previous_client_mode

    @contextmanager
    def use_api_v1(self) -> Generator["InferenceHTTPClient", None, None]:
        """Temporarily use API version 1 for client operations.

        Yields:
            Generator[InferenceHTTPClient, None, None]: The client instance temporarily using API v1.
        """
        previous_client_mode = self.__client_mode
        self.__client_mode = HTTPClientMode.V1
        try:
            yield self
        finally:
            self.__client_mode = previous_client_mode

    def select_model(self, model_id: str) -> "InferenceHTTPClient":
        """Select a model for inference operations.

        Args:
            model_id (str): The identifier of the model to select.

        Returns:
            InferenceHTTPClient: The client instance with the selected model.
        """
        self.__selected_model = model_id
        return self

    @contextmanager
    def use_model(self, model_id: str) -> Generator["InferenceHTTPClient", None, None]:
        """Temporarily use a specific model for inference operations.

        Args:
            model_id (str): The identifier of the model to use.

        Yields:
            Generator[InferenceHTTPClient, None, None]: The client instance temporarily using the specified model.
        """
        previous_model = self.__selected_model
        self.__selected_model = model_id
        try:
            yield self
        finally:
            self.__selected_model = previous_model

    @wrap_errors
    def get_server_info(self) -> ServerInfo:
        """Get information about the inference server.

        Returns:
            ServerInfo: Information about the server configuration and status.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        response = requests.get(f"{self.__api_url}/info")
        response.raise_for_status()
        response_payload = response.json()
        return ServerInfo.from_dict(response_payload)

    def infer_on_stream(
        self,
        input_uri: str,
        model_id: Optional[str] = None,
    ) -> Generator[Tuple[Union[str, int], np.ndarray, dict], None, None]:
        """Run inference on a video stream or sequence of images.

        Args:
            input_uri (str): URI of the input stream or directory.
            model_id (Optional[str], optional): Model identifier to use for inference. Defaults to None.

        Yields:
            Generator[Tuple[Union[str, int], np.ndarray, dict], None, None]: Tuples of (frame reference, frame data, prediction).
        """
        for reference, frame in load_stream_inference_input(
            input_uri=input_uri,
            image_extensions=self.__inference_configuration.image_extensions_for_directory_scan,
        ):
            prediction = self.infer(
                inference_input=frame,
                model_id=model_id,
            )
            yield reference, frame, prediction

    @wrap_errors
    def infer(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Run inference on one or more images.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) for inference.
            model_id (Optional[str], optional): Model identifier to use for inference. Defaults to None.

        Returns:
            Union[dict, List[dict]]: Inference results for the input image(s).

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        if self.__client_mode is HTTPClientMode.V0:
            return self.infer_from_api_v0(
                inference_input=inference_input,
                model_id=model_id,
            )
        return self.infer_from_api_v1(
            inference_input=inference_input,
            model_id=model_id,
        )

    @wrap_errors_async
    async def infer_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Run inference asynchronously on one or more images.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) for inference.
            model_id (Optional[str], optional): Model identifier to use for inference. Defaults to None.

        Returns:
            Union[dict, List[dict]]: Inference results for the input image(s).

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        if self.__client_mode is HTTPClientMode.V0:
            return await self.infer_from_api_v0_async(
                inference_input=inference_input,
                model_id=model_id,
            )
        return await self.infer_from_api_v1_async(
            inference_input=inference_input,
            model_id=model_id,
        )

    def infer_from_api_v0(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Run inference using API v0.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) for inference.
            model_id (Optional[str], optional): Model identifier to use for inference. Defaults to None.

        Returns:
            Union[dict, List[dict]]: Inference results for the input image(s).

        Raises:
            ModelNotSelectedError: If no model is selected.
            APIKeyNotProvided: If API key is required but not provided.
            InvalidModelIdentifier: If the model identifier format is invalid.
        """
        requests_data = self._prepare_infer_from_api_v0_request_data(
            inference_input=inference_input,
            model_id=model_id,
        )
        responses = self._execute_infer_from_api_request(
            requests_data=requests_data,
        )
        results = []
        for request_data, response in zip(requests_data, responses):
            if response_contains_jpeg_image(response=response):
                visualisation = transform_visualisation_bytes(
                    visualisation=response.content,
                    expected_format=self.__inference_configuration.output_visualisation_format,
                )
                parsed_response = {"visualization": visualisation}
            else:
                parsed_response = response.json()
                if parsed_response.get("visualization") is not None:
                    parsed_response["visualization"] = transform_base64_visualisation(
                        visualisation=parsed_response["visualization"],
                        expected_format=self.__inference_configuration.output_visualisation_format,
                    )
            parsed_response = adjust_prediction_to_client_scaling_factor(
                prediction=parsed_response,
                scaling_factor=request_data.image_scaling_factors[0],
            )
            results.append(parsed_response)
        return unwrap_single_element_list(sequence=results)

    def _execute_infer_from_api_request(
        self,
        requests_data: List[RequestData],
    ) -> List[Response]:
        responses = execute_requests_packages(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        return responses

    def _prepare_infer_from_api_v0_request_data(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: Optional[str] = None,
    ) -> List[RequestData]:
        model_id_to_be_used = model_id or self.__selected_model
        _ensure_model_is_selected(model_id=model_id_to_be_used)
        _ensure_api_key_provided(api_key=self.__api_key)
        model_id_to_be_used = resolve_roboflow_model_alias(model_id=model_id_to_be_used)
        model_id_chunks = model_id_to_be_used.split("/")
        if len(model_id_chunks) != 2:
            raise InvalidModelIdentifier(
                f"Invalid model id: {model_id}. Expected format: project_id/model_version_id."
            )
        max_height, max_width = _determine_client_downsizing_parameters(
            client_downsizing_disabled=self.__inference_configuration.client_downsizing_disabled,
            model_description=None,
            default_max_input_size=self.__inference_configuration.default_max_input_size,
        )
        encoded_inference_inputs = load_static_inference_input(
            inference_input=inference_input,
            max_height=max_height,
            max_width=max_width,
        )
        params = {
            "api_key": self.__api_key,
        }
        params.update(self.__inference_configuration.to_legacy_call_parameters())

        execution_id_value = execution_id.get()
        headers = DEFAULT_HEADERS
        if execution_id_value:
            headers = headers.copy()
            headers[EXECUTION_ID_HEADER] = execution_id_value

        requests_data = prepare_requests_data(
            url=f"{self.__api_url}/{model_id_chunks[0]}/{model_id_chunks[1]}",
            encoded_inference_inputs=encoded_inference_inputs,
            headers=headers,
            parameters=params,
            payload=None,
            max_batch_size=1,
            image_placement=ImagePlacement.DATA,
        )
        return requests_data

    async def infer_from_api_v0_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Run inference using API v0 asynchronously.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) for inference.
            model_id (Optional[str], optional): Model identifier to use for inference. Defaults to None.

        Returns:
            Union[dict, List[dict]]: Inference results for the input image(s).

        Raises:
            ModelNotSelectedError: If no model is selected.
            APIKeyNotProvided: If API key is required but not provided.
            InvalidModelIdentifier: If the model identifier format is invalid.
        """
        model_id_to_be_used = model_id or self.__selected_model
        _ensure_model_is_selected(model_id=model_id_to_be_used)
        _ensure_api_key_provided(api_key=self.__api_key)
        model_id_to_be_used = resolve_roboflow_model_alias(model_id=model_id_to_be_used)
        model_id_chunks = model_id_to_be_used.split("/")
        if len(model_id_chunks) != 2:
            raise InvalidModelIdentifier(
                f"Invalid model id: {model_id}. Expected format: project_id/model_version_id."
            )
        max_height, max_width = _determine_client_downsizing_parameters(
            client_downsizing_disabled=self.__inference_configuration.client_downsizing_disabled,
            model_description=None,
            default_max_input_size=self.__inference_configuration.default_max_input_size,
        )
        encoded_inference_inputs = await load_static_inference_input_async(
            inference_input=inference_input,
            max_height=max_height,
            max_width=max_width,
        )
        params = {
            "api_key": self.__api_key,
        }
        params.update(self.__inference_configuration.to_legacy_call_parameters())

        execution_id_value = execution_id.get()
        headers = DEFAULT_HEADERS
        if execution_id_value:
            headers = headers.copy()
            headers[EXECUTION_ID_HEADER] = execution_id_value

        requests_data = prepare_requests_data(
            url=f"{self.__api_url}/{model_id_chunks[0]}/{model_id_chunks[1]}",
            encoded_inference_inputs=encoded_inference_inputs,
            headers=headers,
            parameters=params,
            payload=None,
            max_batch_size=1,
            image_placement=ImagePlacement.DATA,
        )
        responses = await execute_requests_packages_async(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        results = []
        for request_data, response in zip(requests_data, responses):
            if not issubclass(type(response), dict):
                visualisation = transform_visualisation_bytes(
                    visualisation=response,
                    expected_format=self.__inference_configuration.output_visualisation_format,
                )
                parsed_response = {"visualization": visualisation}
            else:
                parsed_response = response
                if parsed_response.get("visualization") is not None:
                    parsed_response["visualization"] = transform_base64_visualisation(
                        visualisation=parsed_response["visualization"],
                        expected_format=self.__inference_configuration.output_visualisation_format,
                    )
            parsed_response = adjust_prediction_to_client_scaling_factor(
                prediction=parsed_response,
                scaling_factor=request_data.image_scaling_factors[0],
            )
            results.append(parsed_response)
        return unwrap_single_element_list(sequence=results)

    def infer_from_api_v1(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        requests_data = self._prepare_infer_from_api_v1_request_data(
            inference_input=inference_input,
            model_id=model_id,
        )
        responses = self._execute_infer_from_api_request(
            requests_data=requests_data,
        )
        results = []
        for request_data, response in zip(requests_data, responses):
            parsed_response = response.json()
            if not issubclass(type(parsed_response), list):
                parsed_response = [parsed_response]
            for parsed_response_element, scaling_factor in zip(
                parsed_response, request_data.image_scaling_factors
            ):
                if parsed_response_element.get("visualization") is not None:
                    parsed_response_element["visualization"] = (
                        transform_base64_visualisation(
                            visualisation=parsed_response_element["visualization"],
                            expected_format=self.__inference_configuration.output_visualisation_format,
                        )
                    )
                parsed_response_element = adjust_prediction_to_client_scaling_factor(
                    prediction=parsed_response_element,
                    scaling_factor=scaling_factor,
                )
                results.append(parsed_response_element)
        return unwrap_single_element_list(sequence=results)

    def _prepare_infer_from_api_v1_request_data(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: Optional[str] = None,
    ) -> List[RequestData]:
        self.__ensure_v1_client_mode()
        model_id_to_be_used = model_id or self.__selected_model
        _ensure_model_is_selected(model_id=model_id_to_be_used)
        model_id_to_be_used = resolve_roboflow_model_alias(model_id=model_id_to_be_used)
        model_description = self.get_model_description(model_id=model_id_to_be_used)
        max_height, max_width = _determine_client_downsizing_parameters(
            client_downsizing_disabled=self.__inference_configuration.client_downsizing_disabled,
            model_description=model_description,
            default_max_input_size=self.__inference_configuration.default_max_input_size,
        )
        if model_description.task_type not in NEW_INFERENCE_ENDPOINTS:
            raise ModelTaskTypeNotSupportedError(
                f"Model task {model_description.task_type} is not supported by API v1 client."
            )
        encoded_inference_inputs = load_static_inference_input(
            inference_input=inference_input,
            max_height=max_height,
            max_width=max_width,
        )
        payload = {
            "api_key": self.__api_key,
            "model_id": model_id_to_be_used,
        }
        endpoint = NEW_INFERENCE_ENDPOINTS[model_description.task_type]
        payload.update(
            self.__inference_configuration.to_api_call_parameters(
                client_mode=self.__client_mode,
                task_type=model_description.task_type,
            )
        )
        requests_data = prepare_requests_data(
            url=f"{self.__api_url}{endpoint}",
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=self.__inference_configuration.max_batch_size,
            image_placement=ImagePlacement.JSON,
        )
        return requests_data

    async def infer_from_api_v1_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        self.__ensure_v1_client_mode()
        model_id_to_be_used = model_id or self.__selected_model
        _ensure_model_is_selected(model_id=model_id_to_be_used)
        model_id_to_be_used = resolve_roboflow_model_alias(model_id=model_id_to_be_used)
        model_description = await self.get_model_description_async(
            model_id=model_id_to_be_used
        )
        max_height, max_width = _determine_client_downsizing_parameters(
            client_downsizing_disabled=self.__inference_configuration.client_downsizing_disabled,
            model_description=model_description,
            default_max_input_size=self.__inference_configuration.default_max_input_size,
        )
        if model_description.task_type not in NEW_INFERENCE_ENDPOINTS:
            raise ModelTaskTypeNotSupportedError(
                f"Model task {model_description.task_type} is not supported by API v1 client."
            )
        encoded_inference_inputs = await load_static_inference_input_async(
            inference_input=inference_input,
            max_height=max_height,
            max_width=max_width,
        )
        payload = {
            "api_key": self.__api_key,
            "model_id": model_id_to_be_used,
        }
        endpoint = NEW_INFERENCE_ENDPOINTS[model_description.task_type]
        payload.update(
            self.__inference_configuration.to_api_call_parameters(
                client_mode=self.__client_mode,
                task_type=model_description.task_type,
            )
        )
        requests_data = prepare_requests_data(
            url=f"{self.__api_url}{endpoint}",
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=self.__inference_configuration.max_batch_size,
            image_placement=ImagePlacement.JSON,
        )
        responses = await execute_requests_packages_async(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        results = []
        for request_data, parsed_response in zip(requests_data, responses):
            if not issubclass(type(parsed_response), list):
                parsed_response = [parsed_response]
            for parsed_response_element, scaling_factor in zip(
                parsed_response, request_data.image_scaling_factors
            ):
                if parsed_response_element.get("visualization") is not None:
                    parsed_response_element["visualization"] = (
                        transform_base64_visualisation(
                            visualisation=parsed_response_element["visualization"],
                            expected_format=self.__inference_configuration.output_visualisation_format,
                        )
                    )
                parsed_response_element = adjust_prediction_to_client_scaling_factor(
                    prediction=parsed_response_element,
                    scaling_factor=scaling_factor,
                )
                results.append(parsed_response_element)
        return unwrap_single_element_list(sequence=results)

    def get_model_description(
        self, model_id: str, allow_loading: bool = True
    ) -> ModelDescription:
        """Get the description of a model.

        Args:
            model_id (str): The identifier of the model.
            allow_loading (bool, optional): Whether to load the model if not already loaded. Defaults to True.

        Returns:
            ModelDescription: Description of the model.

        Raises:
            WrongClientModeError: If not in API v1 mode.
            ModelNotInitializedError: If the model is not initialized and cannot be loaded.
        """
        self.__ensure_v1_client_mode()
        de_aliased_model_id = resolve_roboflow_model_alias(model_id=model_id)
        registered_models = self.list_loaded_models()
        matching_model = filter_model_descriptions(
            descriptions=registered_models.models,
            model_id=de_aliased_model_id,
        )
        if matching_model is None and allow_loading is True:
            registered_models = self.load_model(model_id=de_aliased_model_id)
            matching_model = filter_model_descriptions(
                descriptions=registered_models.models,
                model_id=de_aliased_model_id,
            )
        if matching_model is not None:
            return matching_model
        raise ModelNotInitializedError(
            f"Model {model_id} (de-aliased: {de_aliased_model_id}) is not initialised and cannot "
            f"retrieve its description."
        )

    async def get_model_description_async(
        self, model_id: str, allow_loading: bool = True
    ) -> ModelDescription:
        """Get the description of a model asynchronously.

        Args:
            model_id (str): The identifier of the model.
            allow_loading (bool, optional): Whether to load the model if not already loaded. Defaults to True.

        Returns:
            ModelDescription: Description of the model.

        Raises:
            WrongClientModeError: If not in API v1 mode.
            ModelNotInitializedError: If the model is not initialized and cannot be loaded.
        """
        self.__ensure_v1_client_mode()
        de_aliased_model_id = resolve_roboflow_model_alias(model_id=model_id)
        registered_models = await self.list_loaded_models_async()
        matching_model = filter_model_descriptions(
            descriptions=registered_models.models,
            model_id=de_aliased_model_id,
        )
        if matching_model is None and allow_loading is True:
            registered_models = await self.load_model_async(
                model_id=de_aliased_model_id
            )
            matching_model = filter_model_descriptions(
                descriptions=registered_models.models,
                model_id=de_aliased_model_id,
            )
        if matching_model is not None:
            return matching_model
        raise ModelNotInitializedError(
            f"Model {model_id} (de-aliased: {de_aliased_model_id}) is not initialised and cannot "
            f"retrieve its description."
        )

    @wrap_errors
    def list_loaded_models(self) -> RegisteredModels:
        """List all models currently loaded on the server.

        Returns:
            RegisteredModels: Information about registered models.

        Raises:
            WrongClientModeError: If not in API v1 mode.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        self.__ensure_v1_client_mode()
        response = requests.get(
            f"{self.__api_url}/model/registry?api_key={self.__api_key}"
        )
        response.raise_for_status()
        response_payload = response.json()
        return RegisteredModels.from_dict(response_payload)

    @wrap_errors_async
    async def list_loaded_models_async(self) -> RegisteredModels:
        """List all models currently loaded on the server asynchronously.

        Returns:
            RegisteredModels: Information about registered models.

        Raises:
            WrongClientModeError: If not in API v1 mode.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        self.__ensure_v1_client_mode()
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.__api_url}/model/registry?api_key={self.__api_key}"
            ) as response:
                response.raise_for_status()
                response_payload = await response.json()
                return RegisteredModels.from_dict(response_payload)

    @wrap_errors
    def load_model(
        self, model_id: str, set_as_default: bool = False
    ) -> RegisteredModels:
        """Load a model onto the server.

        Args:
            model_id (str): The identifier of the model to load.
            set_as_default (bool, optional): Whether to set this model as the default. Defaults to False.

        Returns:
            RegisteredModels: Updated information about registered models.

        Raises:
            WrongClientModeError: If not in API v1 mode.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        self.__ensure_v1_client_mode()
        de_aliased_model_id = resolve_roboflow_model_alias(model_id=model_id)
        response = requests.post(
            f"{self.__api_url}/model/add",
            json={
                "model_id": de_aliased_model_id,
                "api_key": self.__api_key,
            },
            headers=DEFAULT_HEADERS,
        )
        response.raise_for_status()
        response_payload = response.json()
        if set_as_default:
            self.__selected_model = de_aliased_model_id
        return RegisteredModels.from_dict(response_payload)

    @wrap_errors_async
    async def load_model_async(
        self, model_id: str, set_as_default: bool = False
    ) -> RegisteredModels:
        """Load a model onto the server asynchronously.

        Args:
            model_id (str): The identifier of the model to load.
            set_as_default (bool, optional): Whether to set this model as the default. Defaults to False.

        Returns:
            RegisteredModels: Updated information about registered models.

        Raises:
            WrongClientModeError: If not in API v1 mode.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        self.__ensure_v1_client_mode()
        de_aliased_model_id = resolve_roboflow_model_alias(model_id=model_id)
        payload = {
            "model_id": de_aliased_model_id,
            "api_key": self.__api_key,
        }
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.__api_url}/model/add",
                json=payload,
                headers=DEFAULT_HEADERS,
            ) as response:
                response.raise_for_status()
                response_payload = await response.json()
        if set_as_default:
            self.__selected_model = de_aliased_model_id
        return RegisteredModels.from_dict(response_payload)

    @wrap_errors
    def unload_model(self, model_id: str) -> RegisteredModels:
        """Unload a model from the server.

        Args:
            model_id (str): The identifier of the model to unload.

        Returns:
            RegisteredModels: Updated information about registered models.

        Raises:
            WrongClientModeError: If not in API v1 mode.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        self.__ensure_v1_client_mode()
        de_aliased_model_id = resolve_roboflow_model_alias(model_id=model_id)
        response = requests.post(
            f"{self.__api_url}/model/remove",
            json={
                "model_id": de_aliased_model_id,
            },
            headers=DEFAULT_HEADERS,
        )
        response.raise_for_status()
        response_payload = response.json()
        if (
            de_aliased_model_id == self.__selected_model
            or model_id == self.__selected_model
        ):
            self.__selected_model = None
        return RegisteredModels.from_dict(response_payload)

    @wrap_errors_async
    async def unload_model_async(self, model_id: str) -> RegisteredModels:
        self.__ensure_v1_client_mode()
        de_aliased_model_id = resolve_roboflow_model_alias(model_id=model_id)
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.__api_url}/model/remove",
                json={
                    "model_id": de_aliased_model_id,
                },
                headers=DEFAULT_HEADERS,
            ) as response:
                response.raise_for_status()
                response_payload = await response.json()
        if (
            de_aliased_model_id == self.__selected_model
            or model_id == self.__selected_model
        ):
            self.__selected_model = None
        return RegisteredModels.from_dict(response_payload)

    @wrap_errors
    def unload_all_models(self) -> RegisteredModels:
        self.__ensure_v1_client_mode()
        response = requests.post(f"{self.__api_url}/model/clear")
        response.raise_for_status()
        response_payload = response.json()
        self.__selected_model = None
        return RegisteredModels.from_dict(response_payload)

    @wrap_errors_async
    async def unload_all_models_async(self) -> RegisteredModels:
        self.__ensure_v1_client_mode()
        async with aiohttp.ClientSession() as session:
            async with session.post(f"{self.__api_url}/model/clear") as response:
                response.raise_for_status()
                response_payload = await response.json()
        self.__selected_model = None
        return RegisteredModels.from_dict(response_payload)

    @wrap_errors
    def ocr_image(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model: str = "doctr",
        version: Optional[str] = None,
        quantize: Optional[bool] = None,
        generate_bounding_boxes: Optional[bool] = None,
        language_codes: Optional[List[str]] = None,
    ) -> Union[dict, List[dict]]:
        """Run OCR on input image(s).

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) for OCR.
            model (str, optional): OCR model to use ('doctr' or 'trocr'). Defaults to "doctr".
            version (Optional[str], optional): Model version to use. Defaults to None.
                For trocr, supported versions are: 'trocr-small-printed', 'trocr-base-printed', 'trocr-large-printed'.
            quantize: (Optional[bool]): flag of EasyOCR to decide which version of model to load
            generate_bounding_boxes: (Optional[bool]): flag of some models (like DocTR) to decide if output variant
                with sv.Detections(...) compatible bounding boxes should be returned (due to historical reasons, some
                old implementations were flattening detected OCR structure into text and were only returning that as
                results).
            language_codes: (Optional[List[str]]): Parameter of EasyOCR that dictates the code of languages that
                model should recognise (leave blank for default for given OCR model version).
        Returns:
            Union[dict, List[dict]]: OCR results for the input image(s).

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        encoded_inference_inputs = load_static_inference_input(
            inference_input=inference_input,
        )
        payload = self.__initialise_payload()
        if version:
            key = f"{model.lower()}_version_id"
            payload[key] = version
        if quantize is not None:
            payload["quantize"] = quantize
        if generate_bounding_boxes is not None:
            payload["generate_bounding_boxes"] = generate_bounding_boxes
        if language_codes is not None:
            payload["language_codes"] = language_codes
        model_path = resolve_ocr_path(model_name=model)
        url = self.__wrap_url_with_api_key(f"{self.__api_url}{model_path}")
        requests_data = prepare_requests_data(
            url=url,
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=1,
            image_placement=ImagePlacement.JSON,
        )
        responses = execute_requests_packages(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        results = [r.json() for r in responses]
        return unwrap_single_element_list(sequence=results)

    @wrap_errors_async
    async def ocr_image_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model: str = "doctr",
        version: Optional[str] = None,
        quantize: Optional[bool] = None,
        generate_bounding_boxes: Optional[bool] = None,
        language_codes: Optional[List[str]] = None,
    ) -> Union[dict, List[dict]]:
        """Run OCR on input image(s) asynchronously.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) for OCR.
            model (str, optional): OCR model to use ('doctr' or 'trocr'). Defaults to "doctr".
            version (Optional[str], optional): Model version to use. Defaults to None.
                For trocr, supported versions are: 'trocr-small-printed', 'trocr-base-printed', 'trocr-large-printed'.
            quantize: (Optional[bool]): flag of EasyOCR to decide which version of model to load
            generate_bounding_boxes: (Optional[bool]): flag of some models (like DocTR) to decide if output variant
                with sv.Detections(...) compatible bounding boxes should be returned (due to historical reasons, some
                old implementations were flattening detected OCR structure into text and were only returning that as
                results).
            language_codes: (Optional[List[str]]): Parameter of EasyOCR that dictates the code of languages that
                model should recognise (leave blank for default for given OCR model version).
        Returns:
            Union[dict, List[dict]]: OCR results for the input image(s).

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        encoded_inference_inputs = await load_static_inference_input_async(
            inference_input=inference_input,
        )
        payload = self.__initialise_payload()
        if version:
            key = f"{model.lower()}_version_id"
            payload[key] = version
        if quantize is not None:
            payload["quantize"] = quantize
        if generate_bounding_boxes is not None:
            payload["generate_bounding_boxes"] = generate_bounding_boxes
        if language_codes is not None:
            payload["language_codes"] = language_codes
        model_path = resolve_ocr_path(model_name=model)
        url = self.__wrap_url_with_api_key(f"{self.__api_url}{model_path}")
        requests_data = prepare_requests_data(
            url=url,
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=1,
            image_placement=ImagePlacement.JSON,
        )
        responses = await execute_requests_packages_async(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        return unwrap_single_element_list(sequence=responses)

    @wrap_errors
    def detect_gazes(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
    ) -> Union[dict, List[dict]]:
        """Detect gazes in input image(s).

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) for gaze detection.

        Returns:
            Union[dict, List[dict]]: Gaze detection results for the input image(s).

        Raises:
            WrongClientModeError: If not in API v1 mode.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        self.__ensure_v1_client_mode()  # Lambda does not support Gaze, so we require v1 mode of client
        result = self._post_images(
            inference_input=inference_input, endpoint="/gaze/gaze_detection"
        )
        return combine_gaze_detections(detections=result)

    @wrap_errors_async
    async def detect_gazes_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
    ) -> Union[dict, List[dict]]:
        """Detect gazes in input image(s) asynchronously.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) for gaze detection.

        Returns:
            Union[dict, List[dict]]: Gaze detection results for the input image(s).

        Raises:
            WrongClientModeError: If not in API v1 mode.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        self.__ensure_v1_client_mode()  # Lambda does not support Gaze, so we require v1 mode of client
        result = await self._post_images_async(
            inference_input=inference_input, endpoint="/gaze/gaze_detection"
        )
        return combine_gaze_detections(detections=result)

    @wrap_errors
    def get_clip_image_embeddings(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        clip_version: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Get CLIP embeddings for input image(s).

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) to embed.
            clip_version (Optional[str], optional): Version of CLIP model to use. Defaults to None.

        Returns:
            Union[dict, List[dict]]: CLIP embeddings for the input image(s).

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        extra_payload = {}
        if clip_version is not None:
            extra_payload["clip_version_id"] = clip_version
        result = self._post_images(
            inference_input=inference_input,
            endpoint="/clip/embed_image",
            extra_payload=extra_payload,
        )
        result = combine_clip_embeddings(embeddings=result)
        return unwrap_single_element_list(result)

    @wrap_errors_async
    async def get_clip_image_embeddings_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        clip_version: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Get CLIP embeddings for input image(s) asynchronously.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s) to embed.
            clip_version (Optional[str], optional): Version of CLIP model to use. Defaults to None.

        Returns:
            Union[dict, List[dict]]: CLIP embeddings for the input image(s).

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        extra_payload = {}
        if clip_version is not None:
            extra_payload["clip_version_id"] = clip_version
        result = await self._post_images_async(
            inference_input=inference_input,
            endpoint="/clip/embed_image",
            extra_payload=extra_payload,
        )
        result = combine_clip_embeddings(embeddings=result)
        return unwrap_single_element_list(result)

    @wrap_errors
    def get_clip_text_embeddings(
        self,
        text: Union[str, List[str]],
        clip_version: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Get CLIP embeddings for input text(s).

        Args:
            text (Union[str, List[str]]): Input text(s) to embed.
            clip_version (Optional[str], optional): Version of CLIP model to use. Defaults to None.

        Returns:
            Union[dict, List[dict]]: CLIP embeddings for the input text(s).

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        payload = self.__initialise_payload()
        payload["text"] = text
        if clip_version is not None:
            payload["clip_version_id"] = clip_version
        headers = DEFAULT_HEADERS.copy()
        execution_id_value = execution_id.get()
        if execution_id_value is not None:
            headers[EXECUTION_ID_HEADER] = execution_id_value

        response = requests.post(
            self.__wrap_url_with_api_key(f"{self.__api_url}/clip/embed_text"),
            json=payload,
            headers=headers,
        )
        _collect_processing_time_from_response(
            response, model_id=clip_version or "clip"
        )
        api_key_safe_raise_for_status(response=response)
        return unwrap_single_element_list(sequence=response.json())

    @wrap_errors_async
    async def get_clip_text_embeddings_async(
        self,
        text: Union[str, List[str]],
        clip_version: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Get CLIP embeddings for input text(s) asynchronously.

        Args:
            text (Union[str, List[str]]): Input text(s) to embed.
            clip_version (Optional[str], optional): Version of CLIP model to use. Defaults to None.

        Returns:
            Union[dict, List[dict]]: CLIP embeddings for the input text(s).

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        payload = self.__initialise_payload()
        payload["text"] = text
        if clip_version is not None:
            payload["clip_version_id"] = clip_version
        async with aiohttp.ClientSession() as session:
            async with session.post(
                self.__wrap_url_with_api_key(f"{self.__api_url}/clip/embed_text"),
                json=payload,
                headers=DEFAULT_HEADERS,
            ) as response:
                response.raise_for_status()
                response_payload = await response.json()
        return unwrap_single_element_list(sequence=response_payload)

    @wrap_errors
    def clip_compare(
        self,
        subject: Union[str, ImagesReference],
        prompt: Union[str, List[str], ImagesReference, List[ImagesReference]],
        subject_type: str = "image",
        prompt_type: str = "text",
        clip_version: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Compare a subject against prompts using CLIP embeddings.

        Args:
            subject (Union[str, ImagesReference]): The subject to compare (image or text).
            prompt (Union[str, List[str], ImagesReference, List[ImagesReference]]): The prompt(s) to compare against.
            subject_type (str, optional): Type of subject ('image' or 'text'). Defaults to "image".
            prompt_type (str, optional): Type of prompt(s) ('image' or 'text'). Defaults to "text".
            clip_version (Optional[str], optional): Version of CLIP model to use. Defaults to None.

        Returns:
            Union[dict, List[dict]]: Comparison results between subject and prompt(s).

        Raises:
            InvalidParameterError: If subject_type or prompt_type is invalid.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        if (
            subject_type not in CLIP_ARGUMENT_TYPES
            or prompt_type not in CLIP_ARGUMENT_TYPES
        ):
            raise InvalidParameterError(
                f"Could not accept `subject_type` and `prompt_type` with values different than {CLIP_ARGUMENT_TYPES}"
            )
        payload = self.__initialise_payload()
        payload["subject_type"] = subject_type
        payload["prompt_type"] = prompt_type
        if clip_version is not None:
            payload["clip_version_id"] = clip_version
        if subject_type == "image":
            encoded_image = load_static_inference_input(
                inference_input=subject,
            )
            payload = inject_images_into_payload(
                payload=payload, encoded_images=encoded_image, key="subject"
            )
        else:
            payload["subject"] = subject
        if prompt_type == "image":
            encoded_inference_inputs = load_static_inference_input(
                inference_input=prompt,
            )
            payload = inject_images_into_payload(
                payload=payload, encoded_images=encoded_inference_inputs, key="prompt"
            )
        else:
            payload["prompt"] = prompt

        headers = DEFAULT_HEADERS.copy()
        execution_id_value = execution_id.get()
        if execution_id_value is not None:
            headers[EXECUTION_ID_HEADER] = execution_id_value

        response = requests.post(
            self.__wrap_url_with_api_key(f"{self.__api_url}/clip/compare"),
            json=payload,
            headers=headers,
        )
        _collect_processing_time_from_response(
            response, model_id=clip_version or "clip"
        )
        api_key_safe_raise_for_status(response=response)
        return response.json()

    @wrap_errors_async
    async def clip_compare_async(
        self,
        subject: Union[str, ImagesReference],
        prompt: Union[str, List[str], ImagesReference, List[ImagesReference]],
        subject_type: str = "image",
        prompt_type: str = "text",
        clip_version: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Compare a subject against prompts using CLIP embeddings asynchronously.

        Args:
            subject (Union[str, ImagesReference]): The subject to compare (image or text).
            prompt (Union[str, List[str], ImagesReference, List[ImagesReference]]): The prompt(s) to compare against.
            subject_type (str, optional): Type of subject ('image' or 'text'). Defaults to "image".
            prompt_type (str, optional): Type of prompt(s) ('image' or 'text'). Defaults to "text".
            clip_version (Optional[str], optional): Version of CLIP model to use. Defaults to None.

        Returns:
            Union[dict, List[dict]]: Comparison results between subject and prompt(s).

        Raises:
            InvalidParameterError: If subject_type or prompt_type is invalid.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        if (
            subject_type not in CLIP_ARGUMENT_TYPES
            or prompt_type not in CLIP_ARGUMENT_TYPES
        ):
            raise InvalidParameterError(
                f"Could not accept `subject_type` and `prompt_type` with values different than {CLIP_ARGUMENT_TYPES}"
            )
        payload = self.__initialise_payload()
        payload["subject_type"] = subject_type
        payload["prompt_type"] = prompt_type
        if clip_version is not None:
            payload["clip_version_id"] = clip_version
        if subject_type == "image":
            encoded_image = await load_static_inference_input_async(
                inference_input=subject,
            )
            payload = inject_images_into_payload(
                payload=payload, encoded_images=encoded_image, key="subject"
            )
        else:
            payload["subject"] = subject
        if prompt_type == "image":
            encoded_inference_inputs = await load_static_inference_input_async(
                inference_input=prompt,
            )
            payload = inject_images_into_payload(
                payload=payload, encoded_images=encoded_inference_inputs, key="prompt"
            )
        else:
            payload["prompt"] = prompt

        async with aiohttp.ClientSession() as session:
            async with session.post(
                self.__wrap_url_with_api_key(f"{self.__api_url}/clip/compare"),
                json=payload,
                headers=DEFAULT_HEADERS,
            ) as response:
                response.raise_for_status()
                return await response.json()

    @wrap_errors
    def get_perception_encoder_image_embeddings(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        perception_encoder_version: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Get Perception Encoder embeddings for input image(s)."""
        extra_payload = {}
        if perception_encoder_version is not None:
            extra_payload["perception_encoder_version_id"] = perception_encoder_version
        result = self._post_images(
            inference_input=inference_input,
            endpoint="/perception_encoder/embed_image",
            extra_payload=extra_payload,
        )
        return unwrap_single_element_list(result)

    @wrap_errors
    def get_perception_encoder_text_embeddings(
        self,
        text: Union[str, List[str]],
        perception_encoder_version: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Get Perception Encoder embeddings for input text(s)."""
        payload = self.__initialise_payload()
        payload["text"] = text
        if perception_encoder_version is not None:
            payload["perception_encoder_version_id"] = perception_encoder_version

        headers = DEFAULT_HEADERS.copy()
        execution_id_value = execution_id.get()
        if execution_id_value is not None:
            headers[EXECUTION_ID_HEADER] = execution_id_value

        response = requests.post(
            self.__wrap_url_with_api_key(
                f"{self.__api_url}/perception_encoder/embed_text"
            ),
            json=payload,
            headers=headers,
        )
        _collect_processing_time_from_response(
            response,
            model_id=perception_encoder_version or "perception_encoder",
        )
        api_key_safe_raise_for_status(response=response)
        return unwrap_single_element_list(sequence=response.json())

    @wrap_errors
    def infer_lmm(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: str,
        prompt: Optional[str] = None,
        model_id_in_path: bool = False,
    ) -> Union[dict, List[dict]]:
        """Run inference using a Large Multimodal Model (LMM).

        This method supports various vision-language models including Florence-2,
        Moondream2, SmolVLM, Qwen2.5-VL, Qwen3-VL, and PaliGemma.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s)
                for inference. Can be file paths, URLs, base64 strings, numpy arrays, or PIL images.
            model_id (str): The identifier of the LMM model to use. Examples include:
                - "florence-2-base", "florence-2-large" for Florence-2
                - "moondream2/moondream2_2b_jul24" for Moondream2
                - "smolvlm2/smolvlm-2.2b-instruct" for SmolVLM
                - "qwen25-vl-7b" for Qwen2.5-VL
                - "qwen3vl-2b-instruct" for Qwen3-VL
            prompt (Optional[str], optional): Text prompt to guide the model. Defaults to None.
            model_id_in_path (bool, optional): If True, includes model_id in the URL path
                (e.g., /infer/lmm/florence-2-base) which enables path-based routing.
                If False (default), model_id is only sent in the request body.

        Returns:
            Union[dict, List[dict]]: Inference results containing the model response.
                The structure depends on the specific model used.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        extra_payload = {"model_id": model_id}
        if prompt is not None:
            extra_payload["prompt"] = prompt

        if model_id_in_path:
            endpoint = f"/infer/lmm/{model_id}"
        else:
            endpoint = "/infer/lmm"

        result = self._post_images(
            inference_input=inference_input,
            endpoint=endpoint,
            extra_payload=extra_payload,
        )
        return result

    @wrap_errors_async
    async def infer_lmm_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: str,
        prompt: Optional[str] = None,
        model_id_in_path: bool = False,
    ) -> Union[dict, List[dict]]:
        """Run inference using a Large Multimodal Model (LMM) asynchronously.

        This method supports various vision-language models including Florence-2,
        Moondream2, SmolVLM, Qwen2.5-VL, Qwen3-VL, and PaliGemma.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s)
                for inference. Can be file paths, URLs, base64 strings, numpy arrays, or PIL images.
            model_id (str): The identifier of the LMM model to use.
            prompt (Optional[str], optional): Text prompt to guide the model. Defaults to None.
            model_id_in_path (bool, optional): If True, includes model_id in the URL path
                (e.g., /infer/lmm/florence-2-base) which enables path-based routing.
                If False (default), model_id is only sent in the request body.

        Returns:
            Union[dict, List[dict]]: Inference results containing the model response.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        extra_payload = {"model_id": model_id}
        if prompt is not None:
            extra_payload["prompt"] = prompt

        if model_id_in_path:
            endpoint = f"/infer/lmm/{model_id}"
        else:
            endpoint = "/infer/lmm"

        result = await self._post_images_async(
            inference_input=inference_input,
            endpoint=endpoint,
            extra_payload=extra_payload,
        )
        return result

    @wrap_errors
    def depth_estimation(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: str = "depth-anything-v3/small",
    ) -> Union[dict, List[dict]]:
        """Run depth estimation on input image(s).

        This method estimates depth maps from images using models like Depth Anything.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s)
                for depth estimation. Can be file paths, URLs, base64 strings, numpy arrays,
                or PIL images.
            model_id (str, optional): The depth estimation model to use. Defaults to
                "depth-anything-v3/small". Supported models include:
                - "depth-anything-v2/small"
                - "depth-anything-v3/small"
                - "depth-anything-v3/base"

        Returns:
            Union[dict, List[dict]]: Depth estimation results containing:
                - normalized_depth: The normalized depth map as a list
                - image: Hex-encoded visualization of the depth map

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        extra_payload = {"model_id": model_id}
        result = self._post_images(
            inference_input=inference_input,
            endpoint="/infer/depth-estimation",
            extra_payload=extra_payload,
        )
        return result

    @wrap_errors_async
    async def depth_estimation_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        model_id: str = "depth-anything-v3/small",
    ) -> Union[dict, List[dict]]:
        """Run depth estimation on input image(s) asynchronously.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s)
                for depth estimation.
            model_id (str, optional): The depth estimation model to use. Defaults to
                "depth-anything-v3/small".

        Returns:
            Union[dict, List[dict]]: Depth estimation results.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        extra_payload = {"model_id": model_id}
        result = await self._post_images_async(
            inference_input=inference_input,
            endpoint="/infer/depth-estimation",
            extra_payload=extra_payload,
        )
        return result

    @wrap_errors
    def sam2_segment_image(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        prompts: Optional[List[dict]] = None,
        sam2_version_id: str = "hiera_tiny",
        multimask_output: bool = True,
        mask_input_format: str = "json",
    ) -> Union[dict, List[dict]]:
        """Run Segment Anything 2 (SAM2) segmentation on input image(s).

        This method performs instance segmentation using SAM2, which can segment
        objects based on point or box prompts.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s)
                for segmentation. Can be file paths, URLs, base64 strings, numpy arrays,
                or PIL images.
            prompts (Optional[List[dict]], optional): List of prompt dictionaries. Each prompt
                can contain:
                - "box": {"x": float, "y": float, "width": float, "height": float}
                - "points": [{"x": float, "y": float, "positive": bool}, ...]
                Defaults to None (automatic segmentation).
            sam2_version_id (str, optional): Version of SAM2 model to use. Options are
                "hiera_large", "hiera_small", "hiera_tiny", "hiera_b_plus".
                Defaults to "hiera_tiny".
            multimask_output (bool, optional): Whether to output multiple masks per prompt.
                Defaults to True.
            mask_input_format (str, optional): Format for mask output. Defaults to "json".

        Returns:
            Union[dict, List[dict]]: Segmentation results containing predictions with masks,
                confidence scores, and bounding boxes.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        extra_payload = {
            "sam2_version_id": sam2_version_id,
            "multimask_output": multimask_output,
            "format": mask_input_format,
        }
        if prompts is not None:
            extra_payload["prompts"] = {"prompts": prompts}
        result = self._post_images(
            inference_input=inference_input,
            endpoint="/sam2/segment_image",
            extra_payload=extra_payload,
        )
        return result

    @wrap_errors_async
    async def sam2_segment_image_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        prompts: Optional[List[dict]] = None,
        sam2_version_id: str = "hiera_tiny",
        multimask_output: bool = True,
        mask_input_format: str = "json",
    ) -> Union[dict, List[dict]]:
        """Run Segment Anything 2 (SAM2) segmentation on input image(s) asynchronously.

        Args:
            inference_input (Union[ImagesReference, List[ImagesReference]]): Input image(s)
                for segmentation.
            prompts (Optional[List[dict]], optional): List of prompt dictionaries.
                Defaults to None.
            sam2_version_id (str, optional): Version of SAM2 model. Defaults to "hiera_tiny".
            multimask_output (bool, optional): Whether to output multiple masks. Defaults to True.
            mask_input_format (str, optional): Format for mask output. Defaults to "json".

        Returns:
            Union[dict, List[dict]]: Segmentation results.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        extra_payload = {
            "sam2_version_id": sam2_version_id,
            "multimask_output": multimask_output,
            "format": mask_input_format,
        }
        if prompts is not None:
            extra_payload["prompts"] = {"prompts": prompts}
        result = await self._post_images_async(
            inference_input=inference_input,
            endpoint="/sam2/segment_image",
            extra_payload=extra_payload,
        )
        return result

    @wrap_errors
    def sam3_3d_infer(
        self,
        inference_input: ImagesReference,
        mask_input: Any,
        model_id: str = "sam3-3d-objects",
        *,
        output_meshes: bool = True,
        output_scene: bool = True,
        with_mesh_postprocess: bool = True,
        with_texture_baking: bool = True,
        use_distillations: bool = False,
    ) -> dict:
        """Generate 3D meshes and Gaussian splatting from a 2D image with mask prompts.

        This method uses SAM3 3D to generate 3D representations from 2D images
        with mask prompts.

        Args:
            inference_input (ImagesReference): Input image for 3D generation.
                Can be a file path, URL, base64 string, numpy array, or PIL image.
            mask_input (Any): Mask input in any supported format:
                - Polygon coordinates: [x1, y1, x2, y2, ...]
                - Binary mask (as numpy array or base64)
                - RLE dictionary
                - List of any of the above for multiple masks
            model_id (str, optional): The SAM3 3D model to use. Defaults to "sam3-3d-objects".
            output_meshes (bool, optional): SAM3 3D always outputs object gaussians, and can
                optionally output object meshes if output_meshes is True. Defaults to True.
            output_scene (bool, optional): Output the combined scene reconstruction in
                addition to individual object reconstructions. Defaults to True.
            with_mesh_postprocess (bool, optional): Enable mesh postprocessing. Defaults to True.
            with_texture_baking (bool, optional): Enable texture baking for meshes. Defaults to True.
            use_distillations (bool, optional): Use the distilled versions of the model components.

        Returns:
            dict: Response containing base64-encoded 3D outputs:
                - mesh_glb: Scene mesh in GLB format (base64 encoded) if output_meshes=True, otherwise None.
                - gaussian_ply: Combined Gaussian splatting in PLY format (base64 encoded)
                - objects: List of individual objects, each containing:
                    - mesh_glb: Object mesh (base64) if output_scene=True and output_meshes=True, otherwise None.
                    - gaussian_ply: Object Gaussian (base64) if output_scene=True, otherwise None.
                    - metadata: {"rotation": [...], "translation": [...], "scale": [...]}
                - time: Inference time in seconds

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        encoded_inference_inputs = load_static_inference_input(
            inference_input=inference_input,
        )
        payload = self.__initialise_payload()
        payload["model_id"] = model_id
        payload["mask_input"] = mask_input
        payload["output_meshes"] = output_meshes
        payload["output_scene"] = output_scene
        payload["with_mesh_postprocess"] = with_mesh_postprocess
        payload["with_texture_baking"] = with_texture_baking
        payload["use_distillations"] = use_distillations

        url = self.__wrap_url_with_api_key(f"{self.__api_url}/sam3_3d/infer")
        requests_data = prepare_requests_data(
            url=url,
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=1,
            image_placement=ImagePlacement.JSON,
        )
        responses = execute_requests_packages(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        return responses[0].json()

    @wrap_errors_async
    async def sam3_3d_infer_async(
        self,
        inference_input: ImagesReference,
        mask_input: Any,
        model_id: str = "sam3-3d-objects",
        *,
        output_meshes: bool = True,
        output_scene: bool = True,
        with_mesh_postprocess: bool = True,
        with_texture_baking: bool = True,
        use_distillations: bool = False,
    ) -> dict:
        """Generate 3D meshes and Gaussian splatting from a 2D image asynchronously.

        Args:
            inference_input (ImagesReference): Input image for 3D generation.
            mask_input (Any): Mask input in any supported format.
            model_id (str, optional): The SAM3 3D model to use. Defaults to "sam3-3d-objects".
            output_meshes (bool, optional): SAM3 3D always outputs object gaussians, and can
                optionally output object meshes if output_meshes is True. Defaults to True.
            output_scene (bool, optional): Output the combined scene reconstruction in
                addition to individual object reconstructions. Defaults to True.
            with_mesh_postprocess (bool, optional): Enable mesh postprocessing. Defaults to True.
            with_texture_baking (bool, optional): Enable texture baking for meshes. Defaults to True.
            use_distillations (bool, optional): Use the distilled versions of the model components.

        Returns:
            dict: Response containing base64-encoded 3D outputs:
                - mesh_glb: Scene mesh in GLB format (base64 encoded) if output_meshes=True, otherwise None.
                - gaussian_ply: Combined Gaussian splatting in PLY format (base64 encoded)
                - objects: List of individual objects, each containing:
                    - mesh_glb: Object mesh (base64) if output_scene=True and output_meshes=True, otherwise None.
                    - gaussian_ply: Object Gaussian (base64) if output_scene=True, otherwise None.
                    - metadata: {"rotation": [...], "translation": [...], "scale": [...]}
                - time: Inference time in seconds

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        encoded_inference_inputs = await load_static_inference_input_async(
            inference_input=inference_input,
        )
        payload = self.__initialise_payload()
        payload["model_id"] = model_id
        payload["mask_input"] = mask_input
        payload["output_meshes"] = output_meshes
        payload["output_scene"] = output_scene
        payload["with_mesh_postprocess"] = with_mesh_postprocess
        payload["with_texture_baking"] = with_texture_baking
        payload["use_distillations"] = use_distillations

        url = self.__wrap_url_with_api_key(f"{self.__api_url}/sam3_3d/infer")
        requests_data = prepare_requests_data(
            url=url,
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=1,
            image_placement=ImagePlacement.JSON,
        )
        responses = await execute_requests_packages_async(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        return responses[0]

    @wrap_errors
    def sam3_concept_segment(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        prompts: List[dict],
        model_id: str = "sam3/sam3_final",
        output_prob_thresh: float = 0.5,
        nms_iou_threshold: Optional[float] = None,
        format: str = "polygon",
    ) -> Union[dict, List[dict]]:
        """Run SAM3 promptable concept segmentation (PCS) on input image(s).

        Performs zero-shot instance segmentation using text or visual prompts.

        Args:
            inference_input: Input image(s) for segmentation.
            prompts: List of prompt dicts, each with keys like "type", "text",
                "output_prob_thresh", "boxes", "box_labels".
            model_id: SAM3 model to use. Defaults to "sam3/sam3_final".
            output_prob_thresh: Global confidence threshold. Defaults to 0.5.
            nms_iou_threshold: IoU threshold for cross-prompt NMS. None disables NMS.
            format: Output mask format, "polygon" or "rle". Defaults to "polygon".

        Returns:
            Segmentation results with prompt_results containing predictions.
        """
        extra_payload = {
            "model_id": model_id,
            "prompts": prompts,
            "output_prob_thresh": output_prob_thresh,
            "format": format,
        }
        if nms_iou_threshold is not None:
            extra_payload["nms_iou_threshold"] = nms_iou_threshold
        return self._post_images(
            inference_input=inference_input,
            endpoint="/sam3/concept_segment",
            extra_payload=extra_payload,
        )

    @wrap_errors_async
    async def sam3_concept_segment_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        prompts: List[dict],
        model_id: str = "sam3/sam3_final",
        output_prob_thresh: float = 0.5,
        nms_iou_threshold: Optional[float] = None,
        format: str = "polygon",
    ) -> Union[dict, List[dict]]:
        """Run SAM3 promptable concept segmentation (PCS) asynchronously.

        Args:
            inference_input: Input image(s) for segmentation.
            prompts: List of prompt dicts.
            model_id: SAM3 model to use. Defaults to "sam3/sam3_final".
            output_prob_thresh: Global confidence threshold. Defaults to 0.5.
            nms_iou_threshold: IoU threshold for cross-prompt NMS. None disables NMS.
            format: Output mask format, "polygon" or "rle". Defaults to "polygon".

        Returns:
            Segmentation results with prompt_results containing predictions.
        """
        extra_payload = {
            "model_id": model_id,
            "prompts": prompts,
            "output_prob_thresh": output_prob_thresh,
            "format": format,
        }
        if nms_iou_threshold is not None:
            extra_payload["nms_iou_threshold"] = nms_iou_threshold
        return await self._post_images_async(
            inference_input=inference_input,
            endpoint="/sam3/concept_segment",
            extra_payload=extra_payload,
        )

    @wrap_errors
    def sam3_visual_segment(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        prompts: Optional[List[dict]] = None,
        multimask_output: bool = True,
        mask_input_format: str = "json",
    ) -> Union[dict, List[dict]]:
        """Run SAM3 promptable visual segmentation (PVS) on input image(s).

        Performs instance segmentation using point or box prompts.

        Args:
            inference_input: Input image(s) for segmentation.
            prompts: List of prompt dicts with "box" and/or "points" keys.
                Defaults to None (automatic segmentation).
            multimask_output: Whether to output multiple masks per prompt.
                Defaults to True.
            mask_input_format: Format for mask output. Defaults to "json".

        Returns:
            Segmentation results containing predictions with masks.
        """
        extra_payload = {
            "multimask_output": multimask_output,
            "format": mask_input_format,
        }
        if prompts is not None:
            extra_payload["prompts"] = {"prompts": prompts}
        return self._post_images(
            inference_input=inference_input,
            endpoint="/sam3/visual_segment",
            extra_payload=extra_payload,
        )

    @wrap_errors_async
    async def sam3_visual_segment_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        prompts: Optional[List[dict]] = None,
        multimask_output: bool = True,
        mask_input_format: str = "json",
    ) -> Union[dict, List[dict]]:
        """Run SAM3 promptable visual segmentation (PVS) asynchronously.

        Args:
            inference_input: Input image(s) for segmentation.
            prompts: List of prompt dicts. Defaults to None.
            multimask_output: Whether to output multiple masks. Defaults to True.
            mask_input_format: Format for mask output. Defaults to "json".

        Returns:
            Segmentation results containing predictions with masks.
        """
        extra_payload = {
            "multimask_output": multimask_output,
            "format": mask_input_format,
        }
        if prompts is not None:
            extra_payload["prompts"] = {"prompts": prompts}
        return await self._post_images_async(
            inference_input=inference_input,
            endpoint="/sam3/visual_segment",
            extra_payload=extra_payload,
        )

    @wrap_errors
    def sam3_embed_image(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        image_id: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Generate SAM3 image embeddings.

        Args:
            inference_input: Input image(s) to embed.
            image_id: Optional cache ID for embeddings. Defaults to None.

        Returns:
            Embedding results with image_id and processing time.
        """
        extra_payload = {}
        if image_id is not None:
            extra_payload["image_id"] = image_id
        return self._post_images(
            inference_input=inference_input,
            endpoint="/sam3/embed_image",
            extra_payload=extra_payload if extra_payload else None,
        )

    @wrap_errors_async
    async def sam3_embed_image_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        image_id: Optional[str] = None,
    ) -> Union[dict, List[dict]]:
        """Generate SAM3 image embeddings asynchronously.

        Args:
            inference_input: Input image(s) to embed.
            image_id: Optional cache ID for embeddings. Defaults to None.

        Returns:
            Embedding results with image_id and processing time.
        """
        extra_payload = {}
        if image_id is not None:
            extra_payload["image_id"] = image_id
        return await self._post_images_async(
            inference_input=inference_input,
            endpoint="/sam3/embed_image",
            extra_payload=extra_payload if extra_payload else None,
        )

    @deprecated(
        reason="Please use run_workflow(...) method. This method will be removed end of Q2 2024"
    )
    @wrap_errors
    def infer_from_workflow(
        self,
        workspace_name: Optional[str] = None,
        workflow_name: Optional[str] = None,
        specification: Optional[dict] = None,
        images: Optional[Dict[str, Any]] = None,
        parameters: Optional[Dict[str, Any]] = None,
        excluded_fields: Optional[List[str]] = None,
        use_cache: bool = True,
        enable_profiling: bool = False,
        workflow_version_id: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """Run inference using a workflow specification.

        Triggers inference from workflow specification at the inference HTTP
        side. Either (`workspace_name` and `workflow_name`) or `workflow_specification` must be
        provided. In the first case - definition of workflow will be fetched
        from Roboflow API, in the latter - `workflow_specification` will be
        used. `images` and `parameters` will be merged into workflow inputs,
        the distinction is made to make sure the SDK can easily serialise
        images and prepare a proper payload. Supported images are numpy arrays,
        PIL.Image and base64 images, links to images and local paths.
        `excluded_fields` will be added to request to filter out results
        of workflow execution at the server side.

        Args:
            workspace_name (Optional[str], optional): Name of the workspace containing the workflow. Defaults to None.
            workflow_name (Optional[str], optional): Name of the workflow. Defaults to None.
            specification (Optional[dict], optional): Direct workflow specification. Defaults to None.
            images (Optional[Dict[str, Any]], optional): Images to process. Defaults to None.
            parameters (Optional[Dict[str, Any]], optional): Additional parameters for the workflow. Defaults to None.
            excluded_fields (Optional[List[str]], optional): Fields to exclude from results. Defaults to None.
            use_cache (bool, optional): Whether to use cached results. Defaults to True.
            enable_profiling (bool, optional): Whether to enable profiling. Defaults to False.

        Returns:
            List[Dict[str, Any]]: Results of the workflow execution.

        Raises:
            InvalidParameterError: If neither workflow identifiers nor specification is provided.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        return self._run_workflow(
            workspace_name=workspace_name,
            workflow_id=workflow_name,
            specification=specification,
            images=images,
            parameters=parameters,
            excluded_fields=excluded_fields,
            legacy_endpoints=True,
            use_cache=use_cache,
            enable_profiling=enable_profiling,
            workflow_version_id=workflow_version_id,
        )

    @wrap_errors
    def run_workflow(
        self,
        workspace_name: Optional[str] = None,
        workflow_id: Optional[str] = None,
        specification: Optional[dict] = None,
        images: Optional[Dict[str, Any]] = None,
        parameters: Optional[Dict[str, Any]] = None,
        excluded_fields: Optional[List[str]] = None,
        use_cache: bool = True,
        enable_profiling: bool = False,
        workflow_version_id: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """Run inference using a workflow specification.

        Triggers inference from workflow specification at the inference HTTP
        side. Either (`workspace_name` and `workflow_id`) or `workflow_specification` must be
        provided. In the first case - definition of workflow will be fetched
        from Roboflow API, in the latter - `workflow_specification` will be
        used. `images` and `parameters` will be merged into workflow inputs,
        the distinction is made to make sure the SDK can easily serialise
        images and prepare a proper payload. Supported images are numpy arrays,
        PIL.Image and base64 images, links to images and local paths.
        `excluded_fields` will be added to request to filter out results
        of workflow execution at the server side.

        **Important!**
        Method is not compatible with inference server <=0.9.18. Please migrate to newer version of
        the server before end of Q2 2024. Until that is done - use old method: infer_from_workflow(...).

        Note:
            Method is not compatible with inference server <=0.9.18. Please migrate to newer version of
            the server before end of Q2 2024. Until that is done - use old method: infer_from_workflow(...).

        Args:
            workspace_name (Optional[str], optional): Name of the workspace containing the workflow. Defaults to None.
            workflow_id (Optional[str], optional): ID of the workflow. Defaults to None.
            specification (Optional[dict], optional): Direct workflow specification. Defaults to None.
            images (Optional[Dict[str, Any]], optional): Images to process. Defaults to None.
            parameters (Optional[Dict[str, Any]], optional): Additional parameters for the workflow. Defaults to None.
            excluded_fields (Optional[List[str]], optional): Fields to exclude from results. Defaults to None.
            use_cache (bool, optional): Whether to use cached results. Defaults to True.
            enable_profiling (bool, optional): Whether to enable profiling. Defaults to False.

        Returns:
            List[Dict[str, Any]]: Results of the workflow execution.

        Raises:
            InvalidParameterError: If neither workflow identifiers nor specification is provided.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        return self._run_workflow(
            workspace_name=workspace_name,
            workflow_id=workflow_id,
            specification=specification,
            images=images,
            parameters=parameters,
            excluded_fields=excluded_fields,
            legacy_endpoints=False,
            use_cache=use_cache,
            enable_profiling=enable_profiling,
            workflow_version_id=workflow_version_id,
        )

    def _run_workflow(
        self,
        workspace_name: Optional[str] = None,
        workflow_id: Optional[str] = None,
        specification: Optional[dict] = None,
        images: Optional[Dict[str, Any]] = None,
        parameters: Optional[Dict[str, Any]] = None,
        excluded_fields: Optional[List[str]] = None,
        legacy_endpoints: bool = False,
        use_cache: bool = True,
        enable_profiling: bool = False,
        workflow_version_id: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        response = self._execute_workflow_request(
            workspace_name=workspace_name,
            workflow_id=workflow_id,
            specification=specification,
            images=images,
            parameters=parameters,
            excluded_fields=excluded_fields,
            legacy_endpoints=legacy_endpoints,
            use_cache=use_cache,
            enable_profiling=enable_profiling,
            workflow_version_id=workflow_version_id,
        )
        response_data = response.json()
        workflow_outputs = response_data["outputs"]
        profiler_trace = response_data.get("profiler_trace", [])
        if enable_profiling:
            save_workflows_profiler_trace(
                directory=self.__inference_configuration.profiling_directory,
                profiler_trace=profiler_trace,
            )
        return decode_workflow_outputs(
            workflow_outputs=workflow_outputs,
            expected_format=self.__inference_configuration.output_visualisation_format,
        )

    def _execute_workflow_request(
        self,
        workspace_name: Optional[str] = None,
        workflow_id: Optional[str] = None,
        specification: Optional[dict] = None,
        images: Optional[Dict[str, Any]] = None,
        parameters: Optional[Dict[str, Any]] = None,
        excluded_fields: Optional[List[str]] = None,
        legacy_endpoints: bool = False,
        use_cache: bool = True,
        enable_profiling: bool = False,
        workflow_version_id: Optional[str] = None,
    ) -> Response:
        named_workflow_specified = (workspace_name is not None) and (
            workflow_id is not None
        )
        if not (named_workflow_specified != (specification is not None)):
            raise InvalidParameterError(
                "Parameters (`workspace_name`, `workflow_id` / `workflow_name`) can be used mutually exclusive with "
                "`specification`, but at least one must be set."
            )
        if images is None:
            images = {}
        if parameters is None:
            parameters = {}
        payload = {
            "api_key": self.__api_key,
            "use_cache": use_cache,
            "enable_profiling": enable_profiling,
        }
        inputs = {}
        for image_name, image in images.items():
            loaded_image = load_nested_batches_of_inference_input(
                inference_input=image,
            )
            inject_nested_batches_of_images_into_payload(
                payload=inputs,
                encoded_images=loaded_image,
                key=image_name,
            )
        inputs.update(parameters)
        payload["inputs"] = inputs
        if excluded_fields is not None:
            payload["excluded_fields"] = excluded_fields
        if specification is not None:
            payload["specification"] = specification
        if specification is not None:
            if legacy_endpoints:
                url = f"{self.__api_url}/infer/workflows"
            else:
                url = f"{self.__api_url}/workflows/run"
        else:
            if workflow_version_id is not None:
                payload["workflow_version_id"] = workflow_version_id
            if legacy_endpoints:
                url = f"{self.__api_url}/infer/workflows/{workspace_name}/{workflow_id}"
            else:
                url = f"{self.__api_url}/{workspace_name}/workflows/{workflow_id}"
        response = send_post_request(
            url=url,
            payload=payload,
            headers=DEFAULT_HEADERS,
            enable_retries=self.__inference_configuration.workflow_run_retries_enabled,
        )
        return response

    @wrap_errors
    def infer_from_yolo_world(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        class_names: List[str],
        model_version: Optional[str] = None,
        confidence: Optional[float] = None,
    ) -> List[dict]:
        """Run inference using YOLO-World model.

        Args:
            inference_input: Input image(s) to run inference on. Can be a single image
                reference or a list of image references.
            class_names: List of class names to detect in the image(s).
            model_version: Optional version of YOLO-World model to use. If not specified,
                uses the default version.
            confidence: Optional confidence threshold for detections. If not specified,
                uses the model's default threshold.

        Returns:
            List of dictionaries containing detection results for each input image.
            Each dictionary contains bounding boxes, class labels, and confidence scores
            for detected objects.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        encoded_inference_inputs = load_static_inference_input(
            inference_input=inference_input,
        )
        payload = self.__initialise_payload()
        payload["text"] = class_names
        if model_version is not None:
            payload["yolo_world_version_id"] = model_version
        if confidence is not None:
            payload["confidence"] = confidence
        url = self.__wrap_url_with_api_key(f"{self.__api_url}/yolo_world/infer")
        requests_data = prepare_requests_data(
            url=url,
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=1,
            image_placement=ImagePlacement.JSON,
        )
        responses = execute_requests_packages(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        return [r.json() for r in responses]

    @wrap_errors_async
    async def infer_from_yolo_world_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        class_names: List[str],
        model_version: Optional[str] = None,
        confidence: Optional[float] = None,
    ) -> List[dict]:
        """Run inference using YOLO-World model asynchronously.

        Args:
            inference_input: Input image(s) to run inference on. Can be a single image
                reference or a list of image references.
            class_names: List of class names to detect in the image(s).
            model_version: Optional version of YOLO-World model to use. If not specified,
                uses the default version.
            confidence: Optional confidence threshold for detections. If not specified,
                uses the model's default threshold.

        Returns:
            List of dictionaries containing detection results for each input image.
            Each dictionary contains bounding boxes, class labels, and confidence scores
            for detected objects.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        encoded_inference_inputs = await load_static_inference_input_async(
            inference_input=inference_input,
        )
        payload = self.__initialise_payload()
        payload["text"] = class_names
        if model_version is not None:
            payload["yolo_world_version_id"] = model_version
        if confidence is not None:
            payload["confidence"] = confidence
        url = self.__wrap_url_with_api_key(f"{self.__api_url}/yolo_world/infer")
        requests_data = prepare_requests_data(
            url=url,
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=1,
            image_placement=ImagePlacement.JSON,
        )
        return await execute_requests_packages_async(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )

    @experimental(
        info="Video processing in inference server is under development. Breaking changes are possible."
    )
    @wrap_errors
    def start_inference_pipeline_with_workflow(
        self,
        video_reference: Union[str, int, List[Union[str, int]]],
        workflow_specification: Optional[dict] = None,
        workspace_name: Optional[str] = None,
        workflow_id: Optional[str] = None,
        image_input_name: str = "image",
        workflows_parameters: Optional[Dict[str, Any]] = None,
        workflows_thread_pool_workers: int = 4,
        cancel_thread_pool_tasks_on_exit: bool = True,
        video_metadata_input_name: str = "video_metadata",
        max_fps: Optional[Union[float, int]] = None,
        source_buffer_filling_strategy: Optional[BufferFillingStrategy] = "DROP_OLDEST",
        source_buffer_consumption_strategy: Optional[
            BufferConsumptionStrategy
        ] = "EAGER",
        video_source_properties: Optional[Dict[str, float]] = None,
        batch_collection_timeout: Optional[float] = None,
        results_buffer_size: int = 64,
    ) -> dict:
        """Starts an inference pipeline using a workflow specification.

        Args:
            video_reference: Path to video file, camera index, or list of video sources.
                Can be a string path, integer camera index, or list of either.
            workflow_specification: Optional workflow specification dictionary. Mutually
                exclusive with workspace_name/workflow_id.
            workspace_name: Optional name of workspace containing workflow. Must be used
                with workflow_id.
            workflow_id: Optional ID of workflow to use. Must be used with workspace_name.
            image_input_name: Name of the image input node in workflow. Defaults to "image".
            workflows_parameters: Optional parameters to pass to workflow.
            workflows_thread_pool_workers: Number of worker threads for workflow execution.
                Defaults to 4.
            cancel_thread_pool_tasks_on_exit: Whether to cancel pending tasks when exiting.
                Defaults to True.
            video_metadata_input_name: Name of video metadata input in workflow.
                Defaults to "video_metadata".
            max_fps: Optional maximum FPS to process video at.
            source_buffer_filling_strategy: Strategy for filling source buffer when full.
                One of: "WAIT", "DROP_OLDEST", "ADAPTIVE_DROP_OLDEST", "DROP_LATEST",
                "ADAPTIVE_DROP_LATEST". Defaults to "DROP_OLDEST".
            source_buffer_consumption_strategy: Strategy for consuming from source buffer.
                One of: "LAZY", "EAGER". Defaults to "EAGER".
            video_source_properties: Optional dictionary of video source properties.
            batch_collection_timeout: Optional timeout for batch collection in seconds.
            results_buffer_size: Size of results buffer. Defaults to 64.

        Returns:
            dict: Response containing pipeline initialization details.

        Raises:
            InvalidParameterError: If workflow specification parameters are invalid.
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        named_workflow_specified = (workspace_name is not None) and (
            workflow_id is not None
        )
        if not (named_workflow_specified != (workflow_specification is not None)):
            raise InvalidParameterError(
                "Parameters (`workspace_name`, `workflow_id`) can be used mutually exclusive with "
                "`workflow_specification`, but at least one must be set."
            )
        payload = {
            "api_key": self.__api_key,
            "video_configuration": {
                "type": "VideoConfiguration",
                "video_reference": video_reference,
                "max_fps": max_fps,
                "source_buffer_filling_strategy": source_buffer_filling_strategy,
                "source_buffer_consumption_strategy": source_buffer_consumption_strategy,
                "video_source_properties": video_source_properties,
                "batch_collection_timeout": batch_collection_timeout,
            },
            "processing_configuration": {
                "type": "WorkflowConfiguration",
                "workflow_specification": workflow_specification,
                "workspace_name": workspace_name,
                "workflow_id": workflow_id,
                "image_input_name": image_input_name,
                "workflows_parameters": workflows_parameters,
                "workflows_thread_pool_workers": workflows_thread_pool_workers,
                "cancel_thread_pool_tasks_on_exit": cancel_thread_pool_tasks_on_exit,
                "video_metadata_input_name": video_metadata_input_name,
            },
            "sink_configuration": {
                "type": "MemorySinkConfiguration",
                "results_buffer_size": results_buffer_size,
            },
        }
        response = requests.post(
            f"{self.__api_url}/inference_pipelines/initialise",
            json=payload,
        )
        response.raise_for_status()
        return response.json()

    @experimental(
        info="Video processing in inference server is under development. Breaking changes are possible."
    )
    @wrap_errors
    def list_inference_pipelines(self) -> List[dict]:
        """Lists all active inference pipelines on the server.

        This method retrieves information about all currently running inference pipelines
        on the server, including their IDs and status.

        Returns:
            List[dict]: A list of dictionaries containing information about each active
                inference pipeline.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
        """
        payload = {"api_key": self.__api_key}
        response = requests.get(
            f"{self.__api_url}/inference_pipelines/list",
            json=payload,
        )
        api_key_safe_raise_for_status(response=response)
        return response.json()

    @experimental(
        info="Video processing in inference server is under development. Breaking changes are possible."
    )
    @wrap_errors
    def get_inference_pipeline_status(self, pipeline_id: str) -> dict:
        """Gets the current status of a specific inference pipeline.

        Args:
            pipeline_id: The unique identifier of the inference pipeline to check.

        Returns:
            dict: A dictionary containing the current status and details of the pipeline.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
            ValueError: If pipeline_id is empty or None.
        """
        self._ensure_pipeline_id_not_empty(pipeline_id=pipeline_id)
        payload = {"api_key": self.__api_key}
        response = requests.get(
            f"{self.__api_url}/inference_pipelines/{pipeline_id}/status",
            json=payload,
        )
        api_key_safe_raise_for_status(response=response)
        return response.json()

    @experimental(
        info="Video processing in inference server is under development. Breaking changes are possible."
    )
    @wrap_errors
    def pause_inference_pipeline(self, pipeline_id: str) -> dict:
        """Pauses a running inference pipeline.

        Sends a request to pause the specified inference pipeline. The pipeline must be
        currently running for this operation to succeed.

        Args:
            pipeline_id: The unique identifier of the inference pipeline to pause.

        Returns:
            dict: A dictionary containing the response from the server about the pause operation.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
            ValueError: If pipeline_id is empty or None.
        """
        self._ensure_pipeline_id_not_empty(pipeline_id=pipeline_id)
        payload = {"api_key": self.__api_key}
        response = requests.post(
            f"{self.__api_url}/inference_pipelines/{pipeline_id}/pause",
            json=payload,
        )
        api_key_safe_raise_for_status(response=response)
        return response.json()

    @experimental(
        info="Video processing in inference server is under development. Breaking changes are possible."
    )
    @wrap_errors
    def resume_inference_pipeline(self, pipeline_id: str) -> dict:
        """Resumes a paused inference pipeline.

        Sends a request to resume the specified inference pipeline. The pipeline must be
        currently paused for this operation to succeed.

        Args:
            pipeline_id: The unique identifier of the inference pipeline to resume.

        Returns:
            dict: A dictionary containing the response from the server about the resume operation.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
            ValueError: If pipeline_id is empty or None.
        """
        self._ensure_pipeline_id_not_empty(pipeline_id=pipeline_id)
        payload = {"api_key": self.__api_key}
        response = requests.post(
            f"{self.__api_url}/inference_pipelines/{pipeline_id}/resume",
            json=payload,
        )
        api_key_safe_raise_for_status(response=response)
        return response.json()

    @experimental(
        info="Video processing in inference server is under development. Breaking changes are possible."
    )
    @wrap_errors
    def terminate_inference_pipeline(self, pipeline_id: str) -> dict:
        """Terminates a running inference pipeline.

        Sends a request to terminate the specified inference pipeline. This will stop all
        processing and free up associated resources.

        Args:
            pipeline_id: The unique identifier of the inference pipeline to terminate.

        Returns:
            dict: A dictionary containing the response from the server about the termination operation.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
            ValueError: If pipeline_id is empty or None.
        """
        self._ensure_pipeline_id_not_empty(pipeline_id=pipeline_id)
        payload = {"api_key": self.__api_key}
        response = requests.post(
            f"{self.__api_url}/inference_pipelines/{pipeline_id}/terminate",
            json=payload,
        )
        api_key_safe_raise_for_status(response=response)
        return response.json()

    @experimental(
        info="Video processing in inference server is under development. Breaking changes are possible."
    )
    @wrap_errors
    def consume_inference_pipeline_result(
        self,
        pipeline_id: str,
        excluded_fields: Optional[List[str]] = None,
    ) -> dict:
        """Consumes and returns the next available result from an inference pipeline.

        Args:
            pipeline_id: The unique identifier of the inference pipeline to consume results from.
            excluded_fields: Optional list of field names to exclude from the result. If None,
                no fields will be excluded.

        Returns:
            dict: A dictionary containing the next available result from the pipeline.

        Raises:
            HTTPCallErrorError: If there is an error in the HTTP call.
            HTTPClientError: If there is an error with the server connection.
            InvalidParameterError: If pipeline_id is empty or None.
        """
        self._ensure_pipeline_id_not_empty(pipeline_id=pipeline_id)
        if excluded_fields is None:
            excluded_fields = []
        payload = {"api_key": self.__api_key, "excluded_fields": excluded_fields}
        response = requests.get(
            f"{self.__api_url}/inference_pipelines/{pipeline_id}/consume",
            json=payload,
        )
        api_key_safe_raise_for_status(response=response)
        return response.json()

    def _ensure_pipeline_id_not_empty(self, pipeline_id: str) -> None:
        if not pipeline_id:
            raise InvalidParameterError("Empty `pipeline_id` parameter detected")

    def _post_images(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        endpoint: str,
        model_id: Optional[str] = None,
        extra_payload: Optional[Dict[str, Any]] = None,
    ) -> Union[dict, List[dict]]:
        encoded_inference_inputs = load_static_inference_input(
            inference_input=inference_input,
        )
        payload = self.__initialise_payload()
        if model_id is not None:
            payload["model_id"] = model_id
        url = self.__wrap_url_with_api_key(f"{self.__api_url}{endpoint}")
        if extra_payload is not None:
            payload.update(extra_payload)
        requests_data = prepare_requests_data(
            url=url,
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=self.__inference_configuration.max_batch_size,
            image_placement=ImagePlacement.JSON,
        )
        responses = execute_requests_packages(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        results = [r.json() for r in responses]
        return unwrap_single_element_list(sequence=results)

    async def _post_images_async(
        self,
        inference_input: Union[ImagesReference, List[ImagesReference]],
        endpoint: str,
        model_id: Optional[str] = None,
        extra_payload: Optional[Dict[str, Any]] = None,
    ) -> Union[dict, List[dict]]:
        encoded_inference_inputs = await load_static_inference_input_async(
            inference_input=inference_input,
        )
        payload = self.__initialise_payload()
        if model_id is not None:
            payload["model_id"] = model_id
        url = self.__wrap_url_with_api_key(f"{self.__api_url}{endpoint}")
        if extra_payload is not None:
            payload.update(extra_payload)
        requests_data = prepare_requests_data(
            url=url,
            encoded_inference_inputs=encoded_inference_inputs,
            headers=DEFAULT_HEADERS,
            parameters=None,
            payload=payload,
            max_batch_size=self.__inference_configuration.max_batch_size,
            image_placement=ImagePlacement.JSON,
        )
        responses = await execute_requests_packages_async(
            requests_data=requests_data,
            request_method=RequestMethod.POST,
            max_concurrent_requests=self.__inference_configuration.max_concurrent_requests,
        )
        return unwrap_single_element_list(sequence=responses)

    def __initialise_payload(self) -> dict:
        if self.__client_mode is not HTTPClientMode.V0:
            return {"api_key": self.__api_key}
        return {}

    def __wrap_url_with_api_key(self, url: str) -> str:
        if self.__client_mode is not HTTPClientMode.V0:
            return url
        return f"{url}?api_key={self.__api_key}"

    def __ensure_v1_client_mode(self) -> None:
        if self.__client_mode is not HTTPClientMode.V1:
            raise WrongClientModeError("Use client mode `v1` to run this operation.")