(task, data)
| 168 | |
| 169 | |
| 170 | def service_data_encoder(task, data): |
| 171 | if CustomEncoder.get(task) is not None: |
| 172 | return CustomEncoder[task](data) |
| 173 | output_keys = TASK_OUTPUTS[task] |
| 174 | result = data |
| 175 | for output_key in output_keys: |
| 176 | if output_key == OutputKeys.OUTPUT_IMG: |
| 177 | result[OutputKeys.OUTPUT_IMG] = encode_array_to_img_base64( |
| 178 | data[OutputKeys.OUTPUT_IMG][..., ::-1]) |
| 179 | elif output_key == OutputKeys.OUTPUT_PCM: |
| 180 | result[OutputKeys.OUTPUT_PCM] = encode_pcm_to_base64( |
| 181 | data[OutputKeys.OUTPUT_PCM]) |
| 182 | result = bytes(json.dumps(result, cls=NumpyEncoder), encoding='utf8') |
| 183 | return result |
nothing calls this directly
no test coverage detected
searching dependent graphs…