feat(ray): add help functions for ray model (#40)

Because

- provide support for constructing compatible `ray` model for `Instill
Model`

This commit

- add helper functions for constructing `ray` model

Resolves INS-2497
Resolves INS-2498

.pydocstyle.ini

@@ -12,5 +12,4 @@ add_select = D211
 # D107: Missing docstring in __init__
 # D202: No blank lines allowed after function docstring
 add_ignore = D100,D101,D102,D103,D104,D105,D107,D202
-
-match_dir = '^(?!protogen)'
+match = '(?!test_|!.*_pb2).*\.py'

.pylint.ini

@@ -11,7 +11,7 @@ ignore=CVS,protogen
 
 # Add files or directories matching the regex patterns to the blacklist. The
 # regex matches against base names, not paths.
-ignore-patterns=
+ignore-patterns=.*_pb2.*
 
 # Python code to execute, usually for sys.path manipulation such as
 # pygtk.require().

instill/configuration/__init__.py

@@ -6,6 +6,9 @@
 import yaml
 from pydantic import BaseModel
 
+CLOUD_RAY_ADDRESS = "ray://core_ray_server:10001"
+CORE_RAY_ADDRESS = "ray://ray_server:10001"
+
 CONFIG_DIR = Path(
     os.getenv(
         "INSTILL_SYSTEM_CONFIG_PATH",

instill/helpers/ray_helper.py

@@ -0,0 +1,146 @@
+import argparse
+import struct
+from enum import Enum
+
+import numpy as np
+
+
+class DataType(Enum):
+    TYPE_BOOL = 1
+    TYPE_UINT8 = 2
+    TYPE_UINT16 = 3
+    TYPE_UINT32 = 4
+    TYPE_UINT64 = 5
+    TYPE_INT8 = 6
+    TYPE_INT16 = 7
+    TYPE_INT32 = 8
+    TYPE_INT64 = 9
+    TYPE_FP16 = 10
+    TYPE_FP32 = 11
+    TYPE_FP64 = 12
+    TYPE_STRING = 13
+
+
+def serialize_byte_tensor(input_tensor):
+    """
+    Serializes a bytes tensor into a flat numpy array of length prepended
+    bytes. The numpy array should use dtype of np.object_. For np.bytes_,
+    numpy will remove trailing zeros at the end of byte sequence and because
+    of this it should be avoided.
+    Parameters
+    ----------
+    input_tensor : np.array
+        The bytes tensor to serialize.
+    Returns
+    -------
+    serialized_bytes_tensor : np.array
+        The 1-D numpy array of type uint8 containing the serialized bytes in 'C' order.
+    Raises
+    ------
+    InferenceServerException
+        If unable to serialize the given tensor.
+    """
+
+    if input_tensor.size == 0:
+        return ()
+
+    # If the input is a tensor of string/bytes objects, then must flatten those
+    # into a 1-dimensional array containing the 4-byte byte size followed by the
+    # actual element bytes. All elements are concatenated together in "C" order.
+    if (input_tensor.dtype == np.object_) or (input_tensor.dtype.type == np.bytes_):
+        flattened_ls: list = []
+        for obj in np.nditer(input_tensor, flags=["refs_ok"], order="C"):
+            # If directly passing bytes to BYTES type,
+            # don't convert it to str as Python will encode the
+            # bytes which may distort the meaning
+            assert isinstance(obj, np.ndarray)
+            if input_tensor.dtype == np.object_:
+                if isinstance(obj.item(), bytes):
+                    s = obj.item()
+                else:
+                    s = str(obj.item()).encode("utf-8")
+            else:
+                s = obj.item()
+            flattened_ls.append(struct.pack("<I", len(s)))
+            flattened_ls.append(s)
+        flattened = b"".join(flattened_ls)
+        return flattened
+    return None
+
+
+def deserialize_bytes_tensor(encoded_tensor):
+    """
+    Deserializes an encoded bytes tensor into an
+    numpy array of dtype of python objects
+
+    Parameters
+    ----------
+    encoded_tensor : bytes
+        The encoded bytes tensor where each element
+        has its length in first 4 bytes followed by
+        the content
+    Returns
+    -------
+    string_tensor : np.array
+        The 1-D numpy array of type object containing the
+        deserialized bytes in 'C' order.
+
+    """
+    strs = []
+    offset = 0
+    val_buf = encoded_tensor
+    while offset < len(val_buf):
+        l = struct.unpack_from("<I", val_buf, offset)[0]
+        offset += 4
+        sb = struct.unpack_from("<{}s".format(l), val_buf, offset)[0]
+        offset += l
+        strs.append(sb)
+    return np.array(strs, dtype=bytes)
+
+
+def deploy_decorator(func):
+    def func_wrapper(*args, **kwargs):
+        num_cpus = float(kwargs.pop("num_cpus", False))
+        num_replicas = int(kwargs.pop("num_replicas", False))
+        og_model_path = kwargs.pop("model_path", False)
+        og_model_string_parts = og_model_path.split("/")
+        application_name = og_model_string_parts[5]
+        model_name = og_model_string_parts[3].split("#")[:2]
+        route_prefix = og_model_string_parts[3].split("#")[3]
+
+        return func(
+            num_cpus=num_cpus,
+            num_replicas=num_replicas,
+            application_name=application_name,
+            model_path=og_model_path,
+            model_name=model_name,
+            route_prefix=route_prefix,
+        )
+
+    return func_wrapper
+
+
+def undeploy_decorator(func):
+    def func_wrapper(*args, **kwargs):
+        og_model_path = kwargs.pop("model_path", False)
+        og_model_string_parts = og_model_path.split("/")
+        model_name = og_model_string_parts[3].split("#")[:2]
+
+        return func(model_name=model_name)
+
+    return func_wrapper
+
+
+def entry():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--func", required=True, choices=["deploy", "undeploy"], help="deploy/undeploy"
+    )
+    parser.add_argument("--model", required=True, help="model path ofr the deployment")
+    parser.add_argument("--cpus", default="0.2", help="num of cpus for this deployment")
+    parser.add_argument(
+        "--replicas", default="1", help="num of replicas for this deployment"
+    )
+    args = parser.parse_args()
+
+    return args

instill/ray_pb2.pyi

@@ -0,0 +1,92 @@
+from google.protobuf.internal import containers as _containers
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from typing import ClassVar as _ClassVar, Iterable as _Iterable, Mapping as _Mapping, Optional as _Optional, Union as _Union
+
+DESCRIPTOR: _descriptor.FileDescriptor
+
+class ModelReadyRequest(_message.Message):
+    __slots__ = ["name", "version"]
+    NAME_FIELD_NUMBER: _ClassVar[int]
+    VERSION_FIELD_NUMBER: _ClassVar[int]
+    name: str
+    version: str
+    def __init__(self, name: _Optional[str] = ..., version: _Optional[str] = ...) -> None: ...
+
+class ModelReadyResponse(_message.Message):
+    __slots__ = ["ready"]
+    READY_FIELD_NUMBER: _ClassVar[int]
+    ready: bool
+    def __init__(self, ready: bool = ...) -> None: ...
+
+class ModelMetadataRequest(_message.Message):
+    __slots__ = ["name", "version"]
+    NAME_FIELD_NUMBER: _ClassVar[int]
+    VERSION_FIELD_NUMBER: _ClassVar[int]
+    name: str
+    version: str
+    def __init__(self, name: _Optional[str] = ..., version: _Optional[str] = ...) -> None: ...
+
+class InferTensor(_message.Message):
+    __slots__ = ["name", "datatype", "shape"]
+    NAME_FIELD_NUMBER: _ClassVar[int]
+    DATATYPE_FIELD_NUMBER: _ClassVar[int]
+    SHAPE_FIELD_NUMBER: _ClassVar[int]
+    name: str
+    datatype: str
+    shape: _containers.RepeatedScalarFieldContainer[int]
+    def __init__(self, name: _Optional[str] = ..., datatype: _Optional[str] = ..., shape: _Optional[_Iterable[int]] = ...) -> None: ...
+
+class ModelMetadataResponse(_message.Message):
+    __slots__ = ["name", "versions", "framework", "inputs", "outputs"]
+    class TensorMetadata(_message.Message):
+        __slots__ = ["name", "datatype", "shape"]
+        NAME_FIELD_NUMBER: _ClassVar[int]
+        DATATYPE_FIELD_NUMBER: _ClassVar[int]
+        SHAPE_FIELD_NUMBER: _ClassVar[int]
+        name: str
+        datatype: str
+        shape: _containers.RepeatedScalarFieldContainer[int]
+        def __init__(self, name: _Optional[str] = ..., datatype: _Optional[str] = ..., shape: _Optional[_Iterable[int]] = ...) -> None: ...
+    NAME_FIELD_NUMBER: _ClassVar[int]
+    VERSIONS_FIELD_NUMBER: _ClassVar[int]
+    FRAMEWORK_FIELD_NUMBER: _ClassVar[int]
+    INPUTS_FIELD_NUMBER: _ClassVar[int]
+    OUTPUTS_FIELD_NUMBER: _ClassVar[int]
+    name: str
+    versions: _containers.RepeatedScalarFieldContainer[str]
+    framework: str
+    inputs: _containers.RepeatedCompositeFieldContainer[ModelMetadataResponse.TensorMetadata]
+    outputs: _containers.RepeatedCompositeFieldContainer[ModelMetadataResponse.TensorMetadata]
+    def __init__(self, name: _Optional[str] = ..., versions: _Optional[_Iterable[str]] = ..., framework: _Optional[str] = ..., inputs: _Optional[_Iterable[_Union[ModelMetadataResponse.TensorMetadata, _Mapping]]] = ..., outputs: _Optional[_Iterable[_Union[ModelMetadataResponse.TensorMetadata, _Mapping]]] = ...) -> None: ...
+
+class ModelInferRequest(_message.Message):
+    __slots__ = ["model_name", "model_version", "inputs", "outputs", "raw_input_contents"]
+    class InferRequestedOutputTensor(_message.Message):
+        __slots__ = ["name"]
+        NAME_FIELD_NUMBER: _ClassVar[int]
+        name: str
+        def __init__(self, name: _Optional[str] = ...) -> None: ...
+    MODEL_NAME_FIELD_NUMBER: _ClassVar[int]
+    MODEL_VERSION_FIELD_NUMBER: _ClassVar[int]
+    INPUTS_FIELD_NUMBER: _ClassVar[int]
+    OUTPUTS_FIELD_NUMBER: _ClassVar[int]
+    RAW_INPUT_CONTENTS_FIELD_NUMBER: _ClassVar[int]
+    model_name: str
+    model_version: str
+    inputs: _containers.RepeatedCompositeFieldContainer[InferTensor]
+    outputs: _containers.RepeatedCompositeFieldContainer[ModelInferRequest.InferRequestedOutputTensor]
+    raw_input_contents: _containers.RepeatedScalarFieldContainer[bytes]
+    def __init__(self, model_name: _Optional[str] = ..., model_version: _Optional[str] = ..., inputs: _Optional[_Iterable[_Union[InferTensor, _Mapping]]] = ..., outputs: _Optional[_Iterable[_Union[ModelInferRequest.InferRequestedOutputTensor, _Mapping]]] = ..., raw_input_contents: _Optional[_Iterable[bytes]] = ...) -> None: ...
+
+class ModelInferResponse(_message.Message):
+    __slots__ = ["model_name", "model_version", "outputs", "raw_output_contents"]
+    MODEL_NAME_FIELD_NUMBER: _ClassVar[int]
+    MODEL_VERSION_FIELD_NUMBER: _ClassVar[int]
+    OUTPUTS_FIELD_NUMBER: _ClassVar[int]
+    RAW_OUTPUT_CONTENTS_FIELD_NUMBER: _ClassVar[int]
+    model_name: str
+    model_version: str
+    outputs: _containers.RepeatedCompositeFieldContainer[InferTensor]
+    raw_output_contents: _containers.RepeatedScalarFieldContainer[bytes]
+    def __init__(self, model_name: _Optional[str] = ..., model_version: _Optional[str] = ..., outputs: _Optional[_Iterable[_Union[InferTensor, _Mapping]]] = ..., raw_output_contents: _Optional[_Iterable[bytes]] = ...) -> None: ...