(args, kwargs)
| 196 | |
| 197 | |
| 198 | def extract_grouped_output_callback_args(args, kwargs): |
| 199 | if "output" in kwargs: |
| 200 | parameters = kwargs["output"] |
| 201 | # Normalize list/tuple of multiple positional outputs to a tuple |
| 202 | if isinstance(parameters, (list, tuple)): |
| 203 | parameters = list(parameters) |
| 204 | |
| 205 | # Make sure dependency grouping contains only Output objects |
| 206 | for dep in flatten_grouping(parameters): |
| 207 | if not isinstance(dep, Output): |
| 208 | raise ValueError( |
| 209 | f"Invalid value provided where an Output dependency " |
| 210 | f"object was expected: {dep}" |
| 211 | ) |
| 212 | |
| 213 | return parameters |
| 214 | |
| 215 | parameters = [] |
| 216 | while args: |
| 217 | next_deps = flatten_grouping(args[0]) |
| 218 | if all(isinstance(d, Output) for d in next_deps): |
| 219 | parameters.append(args.pop(0)) |
| 220 | else: |
| 221 | break |
| 222 | return parameters |
| 223 | |
| 224 | |
| 225 | def extract_grouped_input_state_callback_args_from_kwargs(kwargs): |
no test coverage detected
searching dependent graphs…