code: def python_op_factory(name, schema_name=None):
      class Operator(metaclass=_DaliOperatorMeta):
          def __init__(self, *, device="cpu", **kwargs):
              schema_name = _schema_name(type(self))
              self._spec = _b.OpSpec(schema_name)
              self._schema = _b.GetSchema(schema_name)
  
              # Get the device argument. We will need this to determine the device that our outputs
              # will be stored on. The argument is not listed in schema, so we need to add it
              # manually to spec. TODO(klecki): Make it more generic.
              self._device = device
              self._spec.AddArg("device", self._device)
  
              self._init_args, self._call_args = _separate_kwargs(
                  kwargs
              )
  
              for k in self._call_args.keys():
                  _check_arg_input(self._schema, type(self).__name__, k)
  
              self._preserve = self._init_args.get("preserve", False)
  
              # Process the first part of arguments, due to the historical reasons
              # we need to do it in __init__ for error reporting
              # TODO(klecki): Get rid of two-stage adding of scalar arguments - we can do it
              # but the error message would be worse or delayed.
              # Name is handled in the op instance, keep it for later.
              self._name = self._init_args.pop("name", None)
              _process_arguments(
                  self._schema,
                  self._spec,
                  self._init_args,
                  type(self).__name__,
              )
  
          @property
          def spec(self):
              return self._spec
  
          @property
          def schema(self):
              return self._schema
  
          @property
          def device(self):
              return self._device
  
          @property
          def preserve(self):
              return self._preserve
  
          def __call__(self, *inputs, **kwargs):
              inputs = _preprocess_inputs(
                  inputs,
                  self.__class__.__name__,
                  self._device,
                  self._schema,
              )
              input_sets = _build_input_sets(
                  inputs, self.__class__.__name__
              )
  
              args, arg_inputs = _separate_kwargs(kwargs)
  
              # Due to the fact that we already handled *some* args in init, we need to keep only
              # the new ones.
              args = _resolve_double_definitions(
                  args, self._init_args, keep_old=False
              )
              if self._name is not None:
                  args = _resolve_double_definitions(
                      args, {"name": self._name}
                  )  # restore the name
  
              self._preserve = (
                  self._preserve
                  or args.get("preserve", False)
                  or self._schema.IsNoPrune()
              )
  
              # Adding argument inputs is fully delayed into call, so we just do the check
              arg_inputs = _resolve_double_definitions(
                  arg_inputs, self._call_args
              )
  
              # Create OperatorInstance for every input set.
              # OperatorInstance handles the creation of OpSpec and generation of output DataNodes
              op_instances = []
              for input_set in input_sets:
                  op_instances.append(
                      _OperatorInstance(
                          input_set,
                          arg_inputs,
                          args,
                          self._init_args,
                          self,
                      )
                  )
  
              # Tie the instances together
              relation_id = op_instances[0].id
              for op in op_instances:
                  op.relation_id = relation_id
  
              # If we don't have multiple input sets, flatten the result
              if len(op_instances) == 1:
                  result = op_instances[0].unwrapped_outputs
              else:
                  outputs = []
                  for op in op_instances:
                      outputs.append(op.outputs)
                  result = _repack_output_sets(outputs)
              return result
  
      Operator.__name__ = str(name)
      Operator.schema_name = schema_name or Operator.__name__
      Operator._generated = (
          True  # The class was generated using python_op_factory
      )
      Operator.__call__.__doc__ = _docs._docstring_generator_call(
          Operator.schema_name
      )
      return Operator
  
imports: import sys
  import threading
  import tree
  import warnings
  from itertools import count
  import nvidia.dali.python_function_plugin
  from nvidia.dali import backend
  from nvidia.dali import fn
  from nvidia.dali import internal
  from nvidia.dali.data_node import DataNode
  from nvidia.dali.pipeline import Pipeline
  from nvidia.dali.types import _type_name_convert_to_string, _type_convert_value, _default_converter, _vector_element_type, ScalarConstant, Constant
  from nvidia.dali import _conditionals
  from nvidia.dali.ops import _registry, _names, _docs, _operator_utils
  from nvidia.dali.ops._registry import cpu_ops, mixed_ops, gpu_ops, register_cpu_op, register_gpu_op
  from nvidia.dali.ops._names import _op_name, _process_op_name, _schema_name
  from nvidia.dali.ops._operator_utils import _build_input_sets, _repack_output_sets
  from nvidia.dali.external_source import ExternalSource
  from nvidia.dali.ops._operators.python_function import PythonFunctionBase, PythonFunction, DLTensorPythonFunction, _dlpack_to_array, _dlpack_from_array
  from nvidia.dali.ops._operators.compose import Compose
  from nvidia.dali.ops._operators.math import _arithm_op, _group_inputs, _generate_input_desc
label_desc: Use image node_id 12 for calling a global function with example usage: python_op_factory(name, schema_name) and returns: Operator, self, self, self, self, result