o bi+@sddlmZmZmZmZmZddlmZddlm Z ddl m Z ddl m Z dZe GdddeZe Gd d d eZe Gd d d Zd S))AnyDictListUnionOptional)DAGNodeget_dag_node_str)type_to_string) DeveloperAPI__in_context_manager__c seZdZdZddddeeeeeee feffffddZ de e dee e fd ee e fd ee e ffd d Z d dZdefddZde de fddZde fddZde fddZdeee fde fddZddZddZde fd d!ZZS)" InputNodeaoRay dag node used in DAG building API to mark entrypoints of a DAG. Should only be function or class method. A DAG can have multiple entrypoints, but only one instance of InputNode exists per DAG, shared among all DAGNodes. Example: .. code-block:: m1.forward / \ dag_input ensemble -> dag_output \ / m2.forward In this pipeline, each user input is broadcasted to both m1.forward and m2.forward as first stop of the DAG, and authored like .. code-block:: python import ray @ray.remote class Model: def __init__(self, val): self.val = val def forward(self, input): return self.val * input @ray.remote def combine(a, b): return a + b with InputNode() as dag_input: m1 = Model.bind(1) m2 = Model.bind(2) m1_output = m1.forward.bind(dag_input[0]) m2_output = m2.forward.bind(dag_input.x) ray_dag = combine.bind(m1_output, m2_output) # Pass mix of args and kwargs as input. ray_dag.execute(1, x=2) # 1 sent to m1, 2 sent to m2 # Alternatively user can also pass single data object, list or dict # and access them via list index, object attribute or dict key str. ray_dag.execute(UserDataObject(m1=1, m2=2)) # dag_input.m1, dag_input.m2 ray_dag.execute([1, 2]) # dag_input[0], dag_input[1] ray_dag.execute({"m1": 1, "m2": 2}) # dag_input["m1"], dag_input["m2"] N) input_type_other_args_to_resolvercsnt|dks t|dkrtdi|_||_|dur+t|tr+|dur%i}t||d<tjgii|ddS)aInputNode should only take attributes of validating and converting input data rather than the input data itself. User input should be provided via `ray_dag.execute(user_input)`. Args: input_type: Describes the data type of inputs user will be giving. - if given through singular InputNode: type of InputNode - if given through InputAttributeNodes: map of key -> type Used when deciding what Gradio block to represent the input nodes with. _other_args_to_resolve: Internal only to keep InputNode's execution context throughput pickling, replacement and serialization. User should not use or pass this field. rz-InputNode should not take any args or kwargs.Nresult_type_string)other_args_to_resolve) len ValueErrorinput_attribute_nodesr isinstancetyper super__init__)selfrrargskwargs __class__F/home/ubuntu/.local/lib/python3.10/site-packages/ray/dag/input_node.pyrCs zInputNode.__init__new_args new_kwargs new_optionsnew_other_args_to_resolvecCs t|dS)N)r)r rr r!r"r#rrr _copy_implds zInputNode._copy_implcOs>|sJdt|dkrt|dkr|dSt|i|S)zExecutor of InputNode.zInputNode is a singleton instance that should be only used in context manager for dag building and execution. See the docstring of class InputNode for examples.r)_in_context_managerr DAGInputDatarrrrrr _execute_implms zInputNode._execute_implreturncCs|jrt|jvr dS|jtS)z2Return if InputNode is created in context manager.F)_bound_other_args_to_resolveIN_CONTEXT_MANAGERrrrrr'{s   zInputNode._in_context_managerkeyvalcCs||j|<dS)zqSet field in parent DAGNode attribute that can be resolved in both pickle and JSON serialization Nr,)rr/r0rrr set_contextszInputNode.set_contextcCs t|dS)N __InputNode__rr.rrr__str__s zInputNode.__str__cCs8t|ts Jd||jvrt||d|j|<|j|S)Nz5Please only access dag input attributes with str key. __getattr__)rstrrInputAttributeNoderr/rrrr5s  zInputNode.__getattr__cCsdt|ttfs Jdd}|jdur||jvrt|j|}||jvr-t||d||j|<|j|S)NRPlease only use int index or str as first-level key to access fields of dag input. __getitem__)rr6intrr rr7)rr/rrrrr:s  zInputNode.__getitem__cCs|td|S)NT)r2r-r.rrr __enter__s zInputNode.__enter__cGsdSNr)rrrrr__exit__szInputNode.__exit__cCd|jvr |jdSdSzvGet type of the output of this DAGNode. Generated by ray.experimental.gradio_utils.type_to_string(). rNr1r.rrrget_result_type  zInputNode.get_result_type)__name__ __module__ __qualname____doc__rrrrr;r6rrrr%r*boolr'r2r4r5r:r<r>rA __classcell__rrrrr s29!       r c seZdZdZ ddedeeefdedeffdd Zd e e d e ee fd e ee fd e ee ffd dZ ddZ defddZdefddZedeeeffddZZS)r7a5Represents partial access of user input based on an index (int), object attribute or dict key (str). Examples: .. code-block:: python with InputNode() as dag_input: a = dag_input[0] b = dag_input.x ray_dag = add.bind(a, b) # This makes a = 1 and b = 2 ray_dag.execute(1, x=2) with InputNode() as dag_input: a = dag_input[0] b = dag_input[1] ray_dag = add.bind(a, b) # This makes a = 2 and b = 3 ray_dag.execute(2, 3) # Alternatively, you can input a single object # and the inputs are automatically indexed from the object: # This makes a = 2 and b = 3 ray_dag.execute([2, 3]) Ndag_input_noder/accessor_methodrc s2||_||_||_tgii||||ddS)N)rIr/rJr)_dag_input_node_key_accessor_methodrr)rrIr/rJrrrrrszInputAttributeNode.__init__r r!r"r#cCst|d|d|d|dS)NrIr/rJr)r7r$rrrr%s zInputAttributeNode._copy_implcOsvt|jtr |j|jS|j}t|jtr,|jdkr||jS|jdkr*t||jSdSt|jtr7||jStd)a0Executor of InputAttributeNode. Args and kwargs are to match base class signature, but not in the implementation. All args and kwargs should be resolved and replaced with value in bound_args and bound_kwargs via bottom-up recursion when current node is executed. r:r5r9N) rrKr(rLr6rMgetattrr;r)rrruser_input_python_objectrrrr*s        z InputAttributeNode._execute_implr+cCst|d|jdS)Nz["z"])r rLr.rrrr4szInputAttributeNode.__str__cCr?r@r1r.rrrrArBz"InputAttributeNode.get_result_typecCs|jSr=)rLr.rrrr/&szInputAttributeNode.keyr=)rCrDrErFr rr;r6rrrrr%r*r4rApropertyr/rHrrrrr7s4"      r7c@s2eZdZdZddZdeeefdefddZ dS) r(zIf user passed multiple args and kwargs directly to dag.execute(), we generate this wrapper for all user inputs as one object, accessible via list index or object attribute key. cOst||_||_dSr=)list_args_kwargsr)rrrr2s  zDAGInputData.__init__r/r+cCs0t|tr |j|St|tr|j|Std)Nr9)rr;rRr6rSrr8rrrr:6s    zDAGInputData.__getitem__N) rCrDrErFrrr;r6rr:rrrrr(+sr(N)typingrrrrrray.dagrray.dag.format_utilsr ray.experimental.gradio_utilsr ray.util.annotationsr r-r r7r(rrrrs    -r