QuantizationOptimizationPipeline is a sorted set PPQ Optimization passes. PPQ is designed as a Multi pass Compiler of quantization network. where pass here refers to a traversal through the entire network. Quantizer is going to calling optimization pass from pipeline one by one
| 29 | |
| 30 | |
| 31 | class QuantizationOptimizationPipeline(Container, Iterable): |
| 32 | """QuantizationOptimizationPipeline is a sorted set PPQ Optimization |
| 33 | passes. |
| 34 | |
| 35 | PPQ is designed as a Multi pass Compiler of quantization network. |
| 36 | where pass here refers to a traversal through the entire network. |
| 37 | |
| 38 | Quantizer is going to calling optimization pass from pipeline one by one to |
| 39 | eventually finish network quantization procedure |
| 40 | """ |
| 41 | def __init__(self, passes: List[QuantizationOptimizationPass]): |
| 42 | super().__init__() |
| 43 | self._pipeline = [] |
| 44 | if passes is not None: |
| 45 | for optim in passes: |
| 46 | self.append_optimization_to_pipeline(optim_pass=optim) |
| 47 | |
| 48 | def __len__(self) -> int: |
| 49 | return len(self._pipeline) |
| 50 | |
| 51 | def __contains__(self, __x: QuantizationOptimizationPass) -> bool: |
| 52 | assert isinstance(__x, QuantizationOptimizationPass), \ |
| 53 | f'Quantization Optimization Pipeline object only suppose to contain optimization passes, '\ |
| 54 | f'while you require to check a/an {type(__x)} whether in the optimization list' |
| 55 | return __x in self._pipeline |
| 56 | |
| 57 | def __iter__(self) -> Iterator[QuantizationOptimizationPass]: |
| 58 | return self._pipeline.__iter__() |
| 59 | |
| 60 | def optimize( |
| 61 | self, graph: BaseGraph, verbose: bool = True, **kwargs |
| 62 | ) -> None: |
| 63 | max_name_length = 0 |
| 64 | if len(self._pipeline) > 0: |
| 65 | names = [p.name for p in self._pipeline] |
| 66 | max_name_length = max([len(name) for name in names]) |
| 67 | |
| 68 | for optim_pass in self._pipeline: |
| 69 | if not isinstance(optim_pass, QuantizationOptimizationPass): |
| 70 | raise TypeError(f'Quantization Optimization Pipeline object only suppose to contain optimization passes only, ' |
| 71 | f'while {str(optim_pass)}({type(optim_pass)}) was found.') |
| 72 | |
| 73 | if verbose: |
| 74 | padding_length = abs(max_name_length - len(optim_pass.name)) |
| 75 | print(f'[{time.strftime("%H:%M:%S", time.localtime())}] {optim_pass.name} Running ... ' |
| 76 | + ' ' * padding_length, end='') |
| 77 | |
| 78 | if not isinstance(graph, BaseGraph): |
| 79 | raise TypeError(f'parameter 1 should be an instance of PPQ BaseGraph when calling optim pass, ' |
| 80 | f'however {type(graph)} was given.') |
| 81 | optim_pass.optimize(graph=graph, **kwargs) |
| 82 | if verbose: print(f'Finished.') |
| 83 | |
| 84 | def append_optimization_to_pipeline(self, optim_pass: QuantizationOptimizationPass, at_front:bool = False): |
| 85 | assert isinstance(optim_pass, QuantizationOptimizationPass), \ |
| 86 | f'Quantization Optimization Pipeline object only suppose to contain optimization passes, '\ |
| 87 | f'while we got a/an {type(optim_pass)} in the optimization list' |
| 88 | if not at_front: |
no outgoing calls
no test coverage detected