Source code for jina.enums

"""
Miscellaneous enums used in Jina.

To use these enums in YAML config, following the example below:

.. highlight:: yaml
.. code-block:: yaml

      chunk_idx:
        uses: index/chunk.yml
        parallel: ${{PARALLEL}}
        parallel_type: !PollingType ANY
        # or
        parallel_type: ANY
        # or
        parallel_type: any
"""

from enum import Enum, IntEnum, EnumMeta


[docs]class EnumType(EnumMeta): """The metaclass for BetterEnum.""" def __new__(cls, *args, **kwargs): """Register a new EnumType :param args: args passed to super() :param kwargs: kwargs passed to super() :return: the registry class """ _cls = super().__new__(cls, *args, **kwargs) return cls.register_class(_cls)
[docs] @staticmethod def register_class(cls): """ Register the class for dumping loading. :param cls: Target class. :return: Registered class. """ reg_cls_set = getattr(cls, '_registered_class', set()) if cls.__name__ not in reg_cls_set or getattr(cls, 'force_register', False): reg_cls_set.add(cls.__name__) setattr(cls, '_registered_class', reg_cls_set) from .jaml import JAML JAML.register(cls) return cls
[docs]class BetterEnum(IntEnum, metaclass=EnumType): """The base class of Enum used in Jina.""" def __str__(self): return self.name
[docs] @classmethod def from_string(cls, s: str): """ Parse the enum from a string. :param s: string representation of the enum value :return: enum value """ try: return cls[s.upper()] except KeyError: raise ValueError( f'{s.upper()} is not a valid enum for {cls!r}, must be one of {list(cls)}' )
@classmethod def _to_yaml(cls, representer, data): """Required by :mod:`pyyaml`. .. note:: In principle, this should inherit from :class:`JAMLCompatible` directly, however, this method is too simple and thus replaced the parent method. :param representer: pyyaml representer :param data: enum value :return: yaml representation """ return representer.represent_scalar( 'tag:yaml.org,2002:str', str(data), style='"' ) @classmethod def _from_yaml(cls, constructor, node): """Required by :mod:`pyyaml`. .. note:: In principle, this should inherit from :class:`JAMLCompatible` directly, however, this method is too simple and thus replaced the parent method. :param constructor: unused :param node: node to derive the enum value from :return: enum value """ return cls.from_string(node.value)
[docs]class SchedulerType(BetterEnum): """The enum for Scheduler Type.""" LOAD_BALANCE = 0 #: balance the workload between Peas, faster peas get more work ROUND_ROBIN = 1 # : workload are scheduled round-robin manner to the peas, assuming all peas have uniform processing speed.
[docs]class PollingType(BetterEnum): """The enum for representing the parallel type of peas in a pod.""" ANY = 1 #: one of the shards will receive the message ALL = 2 #: all shards will receive the message, blocked until all done with the message ALL_ASYNC = 3 #: (reserved) all replica will receive the message, but any one of them can return, useful in backup @property def is_push(self) -> bool: """ Check if :class:`PollingType` is using `push` protocol. :return: True if this :class:`PollingType` is using `push` protocol else False. """ return self.value == 1 @property def is_block(self) -> bool: """ Check if :class:`PollingType` is using `block` protocol. :return: True if this :class:`PollingType` is requiring `block` protocol else False. """ return self.value == 2
[docs]class LogVerbosity(BetterEnum): """Verbosity level of the logger.""" DEBUG = 10 INFO = 20 SUCCESS = 25 WARNING = 30 ERROR = 40 CRITICAL = 50
[docs]class SocketType(BetterEnum): """Enums for representing the socket type in a pod.""" PULL_BIND = 0 PULL_CONNECT = 1 PUSH_BIND = 2 PUSH_CONNECT = 3 SUB_BIND = 4 SUB_CONNECT = 5 PUB_BIND = 6 PUB_CONNECT = 7 PAIR_BIND = 8 PAIR_CONNECT = 9 ROUTER_BIND = 10 DEALER_CONNECT = 11 ROUTER_CONNECT = 13 @property def is_bind(self) -> bool: """ Check if this socket is using `bind` protocol. :return: True if this socket is using `bind` protocol else False. """ return self.value % 2 == 0 @property def is_receive(self) -> bool: """ Check if this socket is used for receiving data. :return: True if this socket is used for receiving data else False. """ return self.value in {0, 1, 4, 5} @property def is_pubsub(self): """ Check if this socket is used for publish or subscribe data. :return: True if this socket is used for publish or subscribe data else False. """ return 4 <= self.value <= 7 @property def paired(self) -> 'SocketType': """ Get the paired SocketType. :return: a paired SocketType. """ return { SocketType.PULL_BIND: SocketType.PUSH_CONNECT, SocketType.PULL_CONNECT: SocketType.PUSH_BIND, SocketType.SUB_BIND: SocketType.PUB_CONNECT, SocketType.SUB_CONNECT: SocketType.PUB_BIND, SocketType.PAIR_BIND: SocketType.PAIR_CONNECT, SocketType.PUSH_CONNECT: SocketType.PULL_BIND, SocketType.PUSH_BIND: SocketType.PULL_CONNECT, SocketType.PUB_CONNECT: SocketType.SUB_BIND, SocketType.PUB_BIND: SocketType.SUB_CONNECT, SocketType.PAIR_CONNECT: SocketType.PAIR_BIND, }[self]
[docs]class FlowBuildLevel(BetterEnum): """ The enum for representing a flow's build level. Some :class:`jina.flow.Flow` class functions require certain build level to run. """ EMPTY = 0 #: Nothing is built GRAPH = 1 #: The underlying graph is built, you may visualize the flow RUNNING = 2 #: the graph is started and all pods are running
[docs]class GatewayProtocolType(BetterEnum): """ Gateway communication protocol """ GRPC = 0 HTTP = 1 WEBSOCKET = 2
[docs]class PeaRoleType(BetterEnum): """The enum of a Pea role.""" SINGLETON = 0 HEAD = 1 TAIL = 2 PARALLEL = 3
[docs]class PodRoleType(BetterEnum): """The enum of a Pod role for visualization.""" POD = 0 JOIN = 1 INSPECT = 2 GATEWAY = 3 INSPECT_AUX_PASS = 4 JOIN_INSPECT = 5 @property def is_inspect(self) -> bool: """ If the role is inspect pod related. :return: True if the Pod role is inspect related else False. """ return self.value in {2, 4}
[docs]class RequestType(BetterEnum): """The enum of Client mode.""" DATA = 0 CONTROL = 1
[docs]class CompressAlgo(BetterEnum): """ The enum of Compress algorithms. .. note:: LZ4 requires additional package, to install it use pip install "jina[lz4]" .. seealso:: https://docs.python.org/3/library/archiving.html """ NONE = 0 LZ4 = 1 ZLIB = 2 GZIP = 3 BZ2 = 4 LZMA = 5
[docs]class OnErrorStrategy(BetterEnum): """ The level of error handling. .. warning:: In theory, all methods below do not 100% guarantee the success execution on the sequel flow. If something is wrong in the upstream, it is hard to CARRY this exception and moving forward without ANY side-effect. """ IGNORE = ( 0 #: Ignore it, keep running all Drivers & Executors logics in the sequel flow ) SKIP_HANDLE = 1 #: Skip all Executors in the sequel, only `pre_hook` and `post_hook` are called THROW_EARLY = 2 #: Immediately throw the exception, the sequel flow will not be running at all
[docs]class FlowInspectType(BetterEnum): """Inspect strategy in the flow.""" HANG = 0 # keep them hanging there REMOVE = 1 # remove them in the build COLLECT = 2 # spawn a new pod and collect them before build @property def is_keep(self) -> bool: """ Check if the target is inspected. :return: True if the target is inspected else False. """ return self.value in {0, 2}
[docs]class InfrastructureType(BetterEnum): """Infrastructure where the Flow runs on.""" JINA = 0 # flow is orchestrated by Jina K8S = 1 # flow is orchestrated by kubernetes
[docs]class RemoteAccessType(BetterEnum): """Remote access type when connect to the host.""" SSH = 0 # ssh connection JINAD = 1 # using rest api via jinad
[docs]class BuildTestLevel(BetterEnum): """Test level in :command:`jina hub build`, higher level includes lower levels.""" NONE = 0 # no build test EXECUTOR = 1 # test at executor level, directly use the config yaml POD_NONDOCKER = 2 # test at pod level, directly use the config yaml POD_DOCKER = 3 # test at pod level but pod --uses the built image FLOW = 4 # test at a simple flow
[docs]class DataInputType(BetterEnum): """Data input type in the request generator.""" AUTO = 0 # auto inference the input type from data (!WARN: could be slow as it relies on try-execept) DOCUMENT = 1 # the input is a full document CONTENT = 2 # the input is just the content of the document
[docs]class RuntimeBackendType(BetterEnum): """Type of backend in runtime.""" THREAD = 0 PROCESS = 1
[docs]class ProgressBarStatus(BetterEnum): """Type of the progress bar status.""" WORKING = 0 DONE = 1 CANCELED = 2 ERROR = 3
[docs]class EmbeddingClsType(BetterEnum): """Enums for representing the type of embeddings supported.""" DENSE = 0 SCIPY_COO = 1 SCIPY_CSR = 2 SCIPY_BSR = 3 SCIPY_CSC = 4 TORCH = 5 TF = 6 @property def is_sparse(self) -> bool: """ Check if is of sparse type :return: True if the type is sparse """ return self.value != 0 @property def is_dense(self) -> bool: """ Check if is of dense type :return: True if the type is dense """ return self.value == 0 @property def is_scipy(self) -> bool: """ Check if is of scipy sparse type :return: True is of scipy sparse type """ return self.value in [1, 2, 3, 4] @property def is_torch(self) -> bool: """ Check if is of torch sparse type :return: True is of torch sparse type """ return self.value == 5 @property def is_tf(self) -> bool: """ Check if is of tf sparse type :return: True is of tf sparse type """ return self.value == 6 @property def scipy_cls_type(self) -> str: """ Return the specific scipy class type (coo, csr, csc, bsr) :return: True is of scipy sparse type """ if self.is_scipy: return self.name.split('_')[1].lower() @property def is_scipy_stackable(self) -> bool: """ Return if the specific scipy class is stackable. (BSR and CSC when stacked are converted into COO) :return: True is class is stackable """ return self.value in [1, 2]
[docs]class RemoteWorkspaceState(str, Enum): """Enum representing state of remote workspace created by JinaD""" PENDING = 'PENDING' CREATING = 'CREATING' UPDATING = 'UPDATING' ACTIVE = 'ACTIVE' FAILED = 'FAILED' DELETING = 'DELETING'
[docs]def replace_enum_to_str(obj): """ Transform BetterEnum type into string. :param obj: Target obj. :return: Transformed obj with string type values. """ for k, v in obj.items(): if isinstance(v, dict): obj[k] = replace_enum_to_str(v) elif isinstance(v, BetterEnum): obj[k] = str(v) return obj