| 203 | |
| 204 | |
| 205 | class QuantizeInputs(ExportPass): |
| 206 | def __init__( |
| 207 | self, |
| 208 | edge_program_manager: EdgeProgramManager, |
| 209 | quantized_inputs_idx: Union[Dict[int, Dict[str, Any]], List[int]], |
| 210 | method_name: Optional[str] = None, |
| 211 | exported_program: Optional[ExportedProgram] = None, |
| 212 | ): |
| 213 | super().__init__() |
| 214 | self.edge_program_manager = edge_program_manager |
| 215 | |
| 216 | self.quantized_inputs_idx_dict = {} |
| 217 | if isinstance(quantized_inputs_idx, dict): |
| 218 | self.quantized_inputs_idx_dict = quantized_inputs_idx |
| 219 | else: |
| 220 | for idx in quantized_inputs_idx: |
| 221 | self.quantized_inputs_idx_dict[idx] = None |
| 222 | self.param_prefix_name = method_name |
| 223 | self.exported_program = exported_program |
| 224 | self.quant_args = {} |
| 225 | |
| 226 | def edge_manager_update_quant_config_method(self, idx, quant_args): |
| 227 | if self.edge_program_manager is not None: |
| 228 | if not self.edge_program_manager._config_methods: |
| 229 | self.edge_program_manager._config_methods = {} |
| 230 | |
| 231 | self.edge_program_manager._config_methods[ |
| 232 | get_config_method_name(self.param_prefix_name, "input", idx, "scale") |
| 233 | ] = quant_args[0] |
| 234 | self.edge_program_manager._config_methods[ |
| 235 | get_config_method_name(self.param_prefix_name, "input", idx, "zp") |
| 236 | ] = quant_args[1] |
| 237 | self.edge_program_manager._config_methods[ |
| 238 | get_config_method_name( |
| 239 | self.param_prefix_name, "input", idx, "quant_min" |
| 240 | ) |
| 241 | ] = quant_args[2] |
| 242 | self.edge_program_manager._config_methods[ |
| 243 | get_config_method_name( |
| 244 | self.param_prefix_name, "input", idx, "quant_max" |
| 245 | ) |
| 246 | ] = quant_args[3] |
| 247 | self.edge_program_manager._config_methods[ |
| 248 | get_config_method_name(self.param_prefix_name, "input", idx, "dtype") |
| 249 | ] = scalar_type_enum(quant_args[4]) |
| 250 | |
| 251 | def edge_manager_update_quant_config_methods_all(self): |
| 252 | if self.edge_program_manager is not None: |
| 253 | for idx, val in self.quant_args.items(): |
| 254 | self.edge_manager_update_quant_config_method(idx, val) |
| 255 | |
| 256 | def call(self, graph_module: torch.fx.GraphModule): |
| 257 | for i, qparams in self.quantized_inputs_idx_dict.items(): |
| 258 | exported_program = ( |
| 259 | self.edge_program_manager.exported_program() |
| 260 | if self.edge_program_manager is not None |
| 261 | else self.exported_program |
| 262 | ) |
no outgoing calls