Calls function with given index and preprocessed input, and measures process time. Parameters: fn_index: index of function to call processed_input: preprocessed input to pass to function iterator: iterator to use if function is a generator
(
self,
fn_index: int,
processed_input: list[Any],
iterator: AsyncIterator[Any] | None = None,
requests: routes.Request | list[routes.Request] | None = None,
event_id: str | None = None,
event_data: EventData | None = None,
)
| 1059 | return processed_outputs |
| 1060 | |
| 1061 | async def call_function( |
| 1062 | self, |
| 1063 | fn_index: int, |
| 1064 | processed_input: list[Any], |
| 1065 | iterator: AsyncIterator[Any] | None = None, |
| 1066 | requests: routes.Request | list[routes.Request] | None = None, |
| 1067 | event_id: str | None = None, |
| 1068 | event_data: EventData | None = None, |
| 1069 | ): |
| 1070 | """ |
| 1071 | Calls function with given index and preprocessed input, and measures process time. |
| 1072 | Parameters: |
| 1073 | fn_index: index of function to call |
| 1074 | processed_input: preprocessed input to pass to function |
| 1075 | iterator: iterator to use if function is a generator |
| 1076 | requests: requests to pass to function |
| 1077 | event_id: id of event in queue |
| 1078 | event_data: data associated with event trigger |
| 1079 | """ |
| 1080 | block_fn = self.fns[fn_index] |
| 1081 | assert block_fn.fn, f"function with index {fn_index} not defined." |
| 1082 | is_generating = False |
| 1083 | request = requests[0] if isinstance(requests, list) else requests |
| 1084 | start = time.time() |
| 1085 | fn = utils.get_function_with_locals(block_fn.fn, self, event_id) |
| 1086 | |
| 1087 | if iterator is None: # If not a generator function that has already run |
| 1088 | if block_fn.inputs_as_dict: |
| 1089 | processed_input = [dict(zip(block_fn.inputs, processed_input))] |
| 1090 | |
| 1091 | processed_input, progress_index, _ = special_args( |
| 1092 | block_fn.fn, processed_input, request, event_data |
| 1093 | ) |
| 1094 | progress_tracker = ( |
| 1095 | processed_input[progress_index] if progress_index is not None else None |
| 1096 | ) |
| 1097 | |
| 1098 | if progress_tracker is not None and progress_index is not None: |
| 1099 | progress_tracker, fn = create_tracker( |
| 1100 | self, event_id, fn, progress_tracker.track_tqdm |
| 1101 | ) |
| 1102 | processed_input[progress_index] = progress_tracker |
| 1103 | |
| 1104 | if inspect.iscoroutinefunction(fn): |
| 1105 | prediction = await fn(*processed_input) |
| 1106 | else: |
| 1107 | prediction = await anyio.to_thread.run_sync( |
| 1108 | fn, *processed_input, limiter=self.limiter |
| 1109 | ) |
| 1110 | else: |
| 1111 | prediction = None |
| 1112 | |
| 1113 | if inspect.isgeneratorfunction(fn) or inspect.isasyncgenfunction(fn): |
| 1114 | if not self.enable_queue: |
| 1115 | raise ValueError("Need to enable queue to use generators.") |
| 1116 | try: |
| 1117 | if iterator is None: |
| 1118 | iterator = cast(AsyncIterator[Any], prediction) |