Creates direct operator.
(
self,
op_class,
op_name,
pipe,
source_context,
next_logical_id,
batch_size,
device_id,
seed,
inputs,
kwargs,
)
| 362 | """ |
| 363 | |
| 364 | def __init__( |
| 365 | self, |
| 366 | op_class, |
| 367 | op_name, |
| 368 | pipe, |
| 369 | source_context, |
| 370 | next_logical_id, |
| 371 | batch_size, |
| 372 | device_id, |
| 373 | seed, |
| 374 | inputs, |
| 375 | kwargs, |
| 376 | ): |
| 377 | """Creates direct operator.""" |
| 378 | |
| 379 | self._batch_size = batch_size |
| 380 | self._separate_kwargs(kwargs) |
| 381 | |
| 382 | if op_name == "_arithmetic_generic_op": |
| 383 | inputs = self._init_arithm_op(kwargs["name"], inputs) |
| 384 | |
| 385 | # Save inputs classification for later verification. |
| 386 | self._inputs_classification = [] |
| 387 | |
| 388 | # When using input sets we have to create separate operators for each input. |
| 389 | input_set_len = -1 |
| 390 | for i, input in enumerate(inputs): |
| 391 | classification = _Classification(input, f"Input {i}") |
| 392 | |
| 393 | if isinstance(classification.is_batch, list): |
| 394 | if input_set_len == -1: |
| 395 | input_set_len = len(classification.is_batch) |
| 396 | elif input_set_len != len(classification.is_batch): |
| 397 | raise ValueError( |
| 398 | "All argument lists for Multiple Input Sets used " |
| 399 | f"with operator '{op_name}' must have the same length." |
| 400 | ) |
| 401 | self._inputs_classification.append(classification) |
| 402 | |
| 403 | if _conditionals.conditionals_enabled(): |
| 404 | if input_set_len != -1: |
| 405 | raise ValueError( |
| 406 | "Multiple input sets are not supported with conditional" |
| 407 | " execution (when `enable_conditionals=True`)." |
| 408 | ) |
| 409 | |
| 410 | self.expected_inputs_size = len(inputs) |
| 411 | |
| 412 | if "device" not in self._init_args and len(inputs) > 0: |
| 413 | self._init_args["device"] = self._inputs_classification[0].device |
| 414 | if "seed" not in self._init_args: |
| 415 | self._init_args["seed"] = seed |
| 416 | |
| 417 | self._device = self._init_args.get("device", "cpu") |
| 418 | self._device_id = device_id |
| 419 | self._expected_inputs_size = len(inputs) |
| 420 | self.op_helper = op_class(**self._init_args) |
| 421 | self._op_name = op_name |
nothing calls this directly
no test coverage detected