Spaces:
Running
on
Zero
Running
on
Zero
File size: 5,317 Bytes
df6c67d |
1 2 3 4 5 6 7 8 9 10 11 12 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 75 76 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 |
from time import perf_counter
from typing import Any, List, Tuple, Union
import numpy as np
from inference.core.entities.requests.inference import InferenceRequest
from inference.core.entities.responses.inference import InferenceResponse
from inference.core.models.types import PreprocessReturnMetadata
class BaseInference:
"""General inference class.
This class provides a basic interface for inference tasks.
"""
def infer(self, image: Any, **kwargs) -> Any:
"""Runs inference on given data."""
preproc_image, returned_metadata = self.preprocess(image, **kwargs)
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
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.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")
|