mirror of
https://github.com/clearml/clearml-serving
synced 2025-01-31 10:56:52 +00:00
102 lines
4.3 KiB
Python
102 lines
4.3 KiB
Python
from typing import Any, Optional, List, Callable
|
|
|
|
|
|
# Preprocess class Must be named "Preprocess"
|
|
# No need to inherit or to implement all methods
|
|
class Preprocess(object):
|
|
"""
|
|
Preprocess class Must be named "Preprocess"
|
|
Otherwise there are No limitations, No need to inherit or to implement all methods
|
|
Notice! This is not thread safe! the same instance may be accessed from multiple threads simultaneously
|
|
"""
|
|
|
|
def __init__(self):
|
|
# set internal state, this will be called only once. (i.e. not per request)
|
|
pass
|
|
|
|
def load(self, local_file_name: str) -> Optional[Any]: # noqa
|
|
"""
|
|
Optional: provide loading method for the model
|
|
useful if we need to load a model in a specific way for the prediction engine to work
|
|
:param local_file_name: file name / path to read load the model from
|
|
:return: Object that will be called with .predict() method for inference
|
|
"""
|
|
pass
|
|
|
|
def preprocess(self, body: dict, collect_custom_statistics_fn: Optional[Callable[[dict], None]]) -> Any: # noqa
|
|
"""
|
|
Optional: do something with the request data, return any type of object.
|
|
The returned object will be passed as is to the inference engine
|
|
|
|
:param body: dictionary as recieved from the RestAPI
|
|
:param collect_custom_statistics_fn: Optional, if provided allows to send a custom set of key/values
|
|
to the statictics collector servicd.
|
|
None is passed if statiscs collector is not configured, or if the current request should not be collected
|
|
|
|
Usage example:
|
|
>>> print(body)
|
|
{"x0": 1, "x1": 2}
|
|
>>> if collect_custom_statistics_fn:
|
|
>>> collect_custom_statistics_fn({"x0": 1, "x1": 2})
|
|
|
|
:return: Object to be passed directly to the model inference
|
|
"""
|
|
return body
|
|
|
|
def postprocess(self, data: Any, collect_custom_statistics_fn: Optional[Callable[[dict], None]]) -> dict: # noqa
|
|
"""
|
|
Optional: post process the data returned from the model inference engine
|
|
returned dict will be passed back as the request result as is.
|
|
|
|
:param data: object as recieved from the inference model function
|
|
:param collect_custom_statistics_fn: Optional, if provided allows to send a custom set of key/values
|
|
to the statictics collector servicd.
|
|
None is passed if statiscs collector is not configured, or if the current request should not be collected
|
|
|
|
Usage example:
|
|
>>> if collect_custom_statistics_fn:
|
|
>>> collect_custom_statistics_fn({"y": 1})
|
|
|
|
:return: Dictionary passed directly as the returned result of the RestAPI
|
|
"""
|
|
return data
|
|
|
|
def process(self, data: Any, collect_custom_statistics_fn: Optional[Callable[[dict], None]]) -> Any: # noqa
|
|
"""
|
|
Optional: do something with the actual data, return any type of object.
|
|
The returned object will be passed as is to the postprocess function engine
|
|
|
|
:param data: object as recieved from the preprocessing function
|
|
:param collect_custom_statistics_fn: Optional, if provided allows to send a custom set of key/values
|
|
to the statictics collector servicd.
|
|
None is passed if statiscs collector is not configured, or if the current request should not be collected
|
|
|
|
Usage example:
|
|
>>> if collect_custom_statistics_fn:
|
|
>>> collect_custom_statistics_fn({"type": "classification"})
|
|
|
|
:return: Object to be passed tp the post-processing function
|
|
"""
|
|
return data
|
|
|
|
def send_request( # noqa
|
|
self,
|
|
endpoint: str,
|
|
version: Optional[str] = None,
|
|
data: Optional[dict] = None
|
|
) -> Optional[dict]:
|
|
"""
|
|
NOTICE: This method will be replaced in runtime, by the inference service
|
|
|
|
Helper method to send model inference requests to the inference service itself.
|
|
This is designed to help with model ensemble, model pipelines, etc.
|
|
On request error return None, otherwise the request result data dictionary
|
|
|
|
Usage example:
|
|
|
|
>>> x0, x1 = 1, 2
|
|
>>> result = self.send_request(endpoint="test_model_sklearn", version="1", data={"x0": x0, "x1": x1})
|
|
>>> y = result["y"]
|
|
"""
|
|
pass
|