Promote all scalar values in the inputs tuple into operator-backed DataNodes. This operation needs to be performed first, so we can have less duplicated constant nodes when dealing with multiple input sets. Parameters ---------- inputs : tuple The inputs can contain one
(inputs, op_name, device, schema=None)
| 790 | |
| 791 | |
| 792 | def _preprocess_inputs(inputs, op_name, device, schema=None): |
| 793 | """Promote all scalar values in the inputs tuple into operator-backed DataNodes. |
| 794 | |
| 795 | This operation needs to be performed first, so we can have less duplicated constant nodes |
| 796 | when dealing with multiple input sets. |
| 797 | |
| 798 | Parameters |
| 799 | ---------- |
| 800 | inputs : tuple |
| 801 | The inputs can contain one level nesting of Multiple Input Sets. |
| 802 | """ |
| 803 | if isinstance(inputs, tuple): |
| 804 | inputs = list(inputs) |
| 805 | |
| 806 | if schema and (len(inputs) < schema.MinNumInput() or len(inputs) > schema.MaxNumInput()): |
| 807 | raise ValueError( |
| 808 | f"Operator {op_name} expects " |
| 809 | f"from {schema.MinNumInput()} to {schema.MaxNumInput()} inputs, " |
| 810 | f"but received {len(inputs)}." |
| 811 | ) |
| 812 | |
| 813 | def is_input(x): |
| 814 | if isinstance(x, (_DataNode, nvidia.dali.types.ScalarConstant)): |
| 815 | return True |
| 816 | # One level of nesting for Multiple Input Sets. It must be a List[DataNode/ScalarConstant] |
| 817 | # with at least one DataNode. |
| 818 | return ( |
| 819 | isinstance(x, (list)) |
| 820 | and any(isinstance(y, _DataNode) for y in x) |
| 821 | and all(isinstance(y, (_DataNode, nvidia.dali.types.ScalarConstant)) for y in x) |
| 822 | ) |
| 823 | |
| 824 | def get_input_device(schema, input_idx): |
| 825 | default_input_device = "gpu" if device == "gpu" else "cpu" |
| 826 | if schema: |
| 827 | input_device = schema.GetInputDevice(input_idx, None, default_input_device) |
| 828 | else: |
| 829 | input_device = default_input_device |
| 830 | return input_device or default_input_device |
| 831 | |
| 832 | def _promote_scalar_constant(value, input_device): |
| 833 | """When ScalarConstant is encountered, promote it to a DataNode, otherwise do |
| 834 | a pass-through. |
| 835 | """ |
| 836 | if isinstance(value, _ScalarConstant): |
| 837 | return _instantiate_constant_node(value, input_device) |
| 838 | return value |
| 839 | |
| 840 | for idx, inp in enumerate(inputs): |
| 841 | if not is_input(inp): |
| 842 | try: |
| 843 | inp = _Constant(inp, device=get_input_device(schema, idx)) |
| 844 | except Exception as ex: |
| 845 | raise TypeError( |
| 846 | f"when calling operator `{op_name}`: " |
| 847 | f"expected inputs of type 'DataNode', list of 'DataNode' " |
| 848 | f"or convertible to constant nodes. Received " |
| 849 | f"input `{idx}` of type '{type(inp).__name__}'." |
no test coverage detected