Skip to content

base

BaseInference

General inference class.

This class provides a basic interface for inference tasks.

Source code in inference/core/models/base.py
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class BaseInference:
    """General inference class.

    This class provides a basic interface for inference tasks.
    """

    @usage_collector
    def infer(self, image: Any, **kwargs) -> Any:
        """Runs inference on given data.
        - image:
            can be a BGR numpy array, filepath, InferenceRequestImage, PIL Image, byte-string, etc.
        """
        preproc_image, returned_metadata = self.preprocess(image, **kwargs)
        logger.debug(
            f"Preprocessed input shape: {getattr(preproc_image, 'shape', None)}"
        )
        predicted_arrays = self.predict(preproc_image, **kwargs)
        postprocessed = self.postprocess(predicted_arrays, returned_metadata, **kwargs)

        return postprocessed

    def preprocess(
        self, image: Any, **kwargs
    ) -> Tuple[np.ndarray, PreprocessReturnMetadata]:
        raise NotImplementedError

    def predict(self, img_in: np.ndarray, **kwargs) -> Tuple[np.ndarray, ...]:
        raise NotImplementedError

    def postprocess(
        self,
        predictions: Tuple[np.ndarray, ...],
        preprocess_return_metadata: PreprocessReturnMetadata,
        **kwargs,
    ) -> Any:
        raise NotImplementedError

    def infer_from_request(
        self, request: InferenceRequest
    ) -> Union[InferenceResponse, List[InferenceResponse]]:
        """Runs inference on a request

        Args:
            request (InferenceRequest): The request object.

        Returns:
            Union[CVInferenceResponse, List[CVInferenceResponse]]: The response object(s).

        Raises:
            NotImplementedError: This method must be implemented by a subclass.
        """
        raise NotImplementedError

    def make_response(
        self, *args, **kwargs
    ) -> Union[InferenceResponse, List[InferenceResponse]]:
        """Constructs an object detection response.

        Raises:
            NotImplementedError: This method must be implemented by a subclass.
        """
        raise NotImplementedError

infer(image, **kwargs)

Runs inference on given data. - image: can be a BGR numpy array, filepath, InferenceRequestImage, PIL Image, byte-string, etc.

Source code in inference/core/models/base.py
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@usage_collector
def infer(self, image: Any, **kwargs) -> Any:
    """Runs inference on given data.
    - image:
        can be a BGR numpy array, filepath, InferenceRequestImage, PIL Image, byte-string, etc.
    """
    preproc_image, returned_metadata = self.preprocess(image, **kwargs)
    logger.debug(
        f"Preprocessed input shape: {getattr(preproc_image, 'shape', None)}"
    )
    predicted_arrays = self.predict(preproc_image, **kwargs)
    postprocessed = self.postprocess(predicted_arrays, returned_metadata, **kwargs)

    return postprocessed

infer_from_request(request)

Runs inference on a request

Parameters:

Name Type Description Default
request InferenceRequest

The request object.

required

Returns:

Type Description
Union[InferenceResponse, List[InferenceResponse]]

Union[CVInferenceResponse, List[CVInferenceResponse]]: The response object(s).

Raises:

Type Description
NotImplementedError

This method must be implemented by a subclass.

Source code in inference/core/models/base.py
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def infer_from_request(
    self, request: InferenceRequest
) -> Union[InferenceResponse, List[InferenceResponse]]:
    """Runs inference on a request

    Args:
        request (InferenceRequest): The request object.

    Returns:
        Union[CVInferenceResponse, List[CVInferenceResponse]]: The response object(s).

    Raises:
        NotImplementedError: This method must be implemented by a subclass.
    """
    raise NotImplementedError

make_response(*args, **kwargs)

Constructs an object detection response.

Raises:

Type Description
NotImplementedError

This method must be implemented by a subclass.

Source code in inference/core/models/base.py
66
67
68
69
70
71
72
73
74
def make_response(
    self, *args, **kwargs
) -> Union[InferenceResponse, List[InferenceResponse]]:
    """Constructs an object detection response.

    Raises:
        NotImplementedError: This method must be implemented by a subclass.
    """
    raise NotImplementedError

Model

Bases: BaseInference

Base Inference Model (Inherits from BaseInference to define the needed methods)

This class provides the foundational methods for inference and logging, and can be extended by specific models.

Methods:

Name Description
log

Print the given message.

clear_cache

Clears any cache if necessary.

Source code in inference/core/models/base.py
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class Model(BaseInference):
    """Base Inference Model (Inherits from BaseInference to define the needed methods)

    This class provides the foundational methods for inference and logging, and can be extended by specific models.

    Methods:
        log(m): Print the given message.
        clear_cache(): Clears any cache if necessary.
    """

    def log(self, m):
        """Prints the given message.

        Args:
            m (str): The message to print.
        """
        print(m)

    def clear_cache(self):
        """Clears any cache if necessary. This method should be implemented in derived classes as needed."""
        pass

    def infer_from_request(
        self,
        request: InferenceRequest,
    ) -> Union[List[InferenceResponse], InferenceResponse]:
        """
        Perform inference based on the details provided in the request, and return the associated responses.
        The function can handle both single and multiple image inference requests. Optionally, it also provides
        a visualization of the predictions if requested.

        Args:
            request (InferenceRequest): The request object containing details for inference, such as the image or
                images to process, any classes to filter by, and whether or not to visualize the predictions.

        Returns:
            Union[List[InferenceResponse], InferenceResponse]: A list of response objects if the request contains
            multiple images, or a single response object if the request contains one image. Each response object
            contains details about the segmented instances, the time taken for inference, and optionally, a visualization.

        Examples:
            >>> request = InferenceRequest(image=my_image, visualize_predictions=True)
            >>> response = infer_from_request(request)
            >>> print(response.time)  # Prints the time taken for inference
            0.125
            >>> print(response.visualization)  # Accesses the visualization of the prediction if available

        Notes:
            - The processing time for each response is included within the response itself.
            - If `visualize_predictions` is set to True in the request, a visualization of the prediction
              is also included in the response.
        """
        t1 = perf_counter()
        responses = self.infer(**request.dict(), return_image_dims=False)
        for response in responses:
            response.time = perf_counter() - t1
            if request.id:
                response.inference_id = request.id

        if request.visualize_predictions:
            for response in responses:
                response.visualization = self.draw_predictions(request, response)

        if not isinstance(request.image, list) and len(responses) > 0:
            responses = responses[0]

        return responses

    def make_response(
        self, *args, **kwargs
    ) -> Union[InferenceResponse, List[InferenceResponse]]:
        """Makes an inference response from the given arguments.

        Args:
            *args: Variable length argument list.
            **kwargs: Arbitrary keyword arguments.

        Returns:
            InferenceResponse: The inference response.
        """
        raise NotImplementedError(self.__class__.__name__ + ".make_response")

clear_cache()

Clears any cache if necessary. This method should be implemented in derived classes as needed.

Source code in inference/core/models/base.py
95
96
97
def clear_cache(self):
    """Clears any cache if necessary. This method should be implemented in derived classes as needed."""
    pass

infer_from_request(request)

Perform inference based on the details provided in the request, and return the associated responses. The function can handle both single and multiple image inference requests. Optionally, it also provides a visualization of the predictions if requested.

Parameters:

Name Type Description Default
request InferenceRequest

The request object containing details for inference, such as the image or images to process, any classes to filter by, and whether or not to visualize the predictions.

required

Returns:

Type Description
Union[List[InferenceResponse], InferenceResponse]

Union[List[InferenceResponse], InferenceResponse]: A list of response objects if the request contains

Union[List[InferenceResponse], InferenceResponse]

multiple images, or a single response object if the request contains one image. Each response object

Union[List[InferenceResponse], InferenceResponse]

contains details about the segmented instances, the time taken for inference, and optionally, a visualization.

Examples:

>>> request = InferenceRequest(image=my_image, visualize_predictions=True)
>>> response = infer_from_request(request)
>>> print(response.time)  # Prints the time taken for inference
0.125
>>> print(response.visualization)  # Accesses the visualization of the prediction if available
Notes
  • The processing time for each response is included within the response itself.
  • If visualize_predictions is set to True in the request, a visualization of the prediction is also included in the response.
Source code in inference/core/models/base.py
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def infer_from_request(
    self,
    request: InferenceRequest,
) -> Union[List[InferenceResponse], InferenceResponse]:
    """
    Perform inference based on the details provided in the request, and return the associated responses.
    The function can handle both single and multiple image inference requests. Optionally, it also provides
    a visualization of the predictions if requested.

    Args:
        request (InferenceRequest): The request object containing details for inference, such as the image or
            images to process, any classes to filter by, and whether or not to visualize the predictions.

    Returns:
        Union[List[InferenceResponse], InferenceResponse]: A list of response objects if the request contains
        multiple images, or a single response object if the request contains one image. Each response object
        contains details about the segmented instances, the time taken for inference, and optionally, a visualization.

    Examples:
        >>> request = InferenceRequest(image=my_image, visualize_predictions=True)
        >>> response = infer_from_request(request)
        >>> print(response.time)  # Prints the time taken for inference
        0.125
        >>> print(response.visualization)  # Accesses the visualization of the prediction if available

    Notes:
        - The processing time for each response is included within the response itself.
        - If `visualize_predictions` is set to True in the request, a visualization of the prediction
          is also included in the response.
    """
    t1 = perf_counter()
    responses = self.infer(**request.dict(), return_image_dims=False)
    for response in responses:
        response.time = perf_counter() - t1
        if request.id:
            response.inference_id = request.id

    if request.visualize_predictions:
        for response in responses:
            response.visualization = self.draw_predictions(request, response)

    if not isinstance(request.image, list) and len(responses) > 0:
        responses = responses[0]

    return responses

log(m)

Prints the given message.

Parameters:

Name Type Description Default
m str

The message to print.

required
Source code in inference/core/models/base.py
87
88
89
90
91
92
93
def log(self, m):
    """Prints the given message.

    Args:
        m (str): The message to print.
    """
    print(m)

make_response(*args, **kwargs)

Makes an inference response from the given arguments.

Parameters:

Name Type Description Default
*args

Variable length argument list.

()
**kwargs

Arbitrary keyword arguments.

{}

Returns:

Name Type Description
InferenceResponse Union[InferenceResponse, List[InferenceResponse]]

The inference response.

Source code in inference/core/models/base.py
145
146
147
148
149
150
151
152
153
154
155
156
157
def make_response(
    self, *args, **kwargs
) -> Union[InferenceResponse, List[InferenceResponse]]:
    """Makes an inference response from the given arguments.

    Args:
        *args: Variable length argument list.
        **kwargs: Arbitrary keyword arguments.

    Returns:
        InferenceResponse: The inference response.
    """
    raise NotImplementedError(self.__class__.__name__ + ".make_response")