| 157 | |
| 158 | |
| 159 | class Operation(OperationBase, Serializable): |
| 160 | def __init__( |
| 161 | self, name: str, op_type: str, |
| 162 | attributes: Dict[str, Any], platform: TargetPlatform = TargetPlatform.UNSPECIFIED, |
| 163 | inputs: List[Variable] = None, outputs: List[Variable] = None, opset: Opset = None) -> None: |
| 164 | OperationBase.__init__(self, name, op_type, attributes, platform=platform, opset=opset) |
| 165 | Serializable.__init__(self) |
| 166 | self._input_vars = [] if inputs is None else inputs |
| 167 | self._output_vars = [] if outputs is None else outputs |
| 168 | |
| 169 | @ property |
| 170 | def socket(self) -> OpSocket: |
| 171 | if self.type in DEFAULT_SOCKET_TABLE: |
| 172 | return DEFAULT_SOCKET_TABLE[self.type](self) |
| 173 | else: |
| 174 | return DEFAULT_SOCKET_CREATOR(self) |
| 175 | |
| 176 | @ property |
| 177 | def inputs(self) -> List[Variable]: |
| 178 | return self._input_vars |
| 179 | |
| 180 | @ property |
| 181 | def outputs(self) -> List[Variable]: |
| 182 | return self._output_vars |
| 183 | |
| 184 | @ property |
| 185 | def parameters(self) -> List[Variable]: |
| 186 | return [var for var in self.inputs if var.is_parameter] |
| 187 | |
| 188 | @ property |
| 189 | def is_linear_activation(self) -> bool: |
| 190 | return self.type in LINEAR_ACTIVATIONS |
| 191 | |
| 192 | @ property |
| 193 | def num_of_parameter(self) -> int: |
| 194 | return len([var for var in self.inputs if var.is_parameter]) |
| 195 | |
| 196 | @ property |
| 197 | def is_boundary(self) -> bool: |
| 198 | up_ops, down_ops = [], [] |
| 199 | for var in self.inputs: |
| 200 | up_ops.append(var.source_op) |
| 201 | for var in self.outputs: |
| 202 | down_ops.extend(var.dest_ops) |
| 203 | return all([op is None for op in up_ops]) or len(down_ops) == 0 |
| 204 | |
| 205 | def __hash__(self) -> int: |
| 206 | return self._name.__hash__() |
| 207 | |
| 208 | def __str__(self) -> str: |
| 209 | return f'{self._name}(Type: {self.type}, Num of Input: {self.num_of_input}, Num of Output: {self.num_of_output})' |
| 210 | |
| 211 | def __repr__(self) -> str: |
| 212 | return f'{self._name}(Type: {self.type}, Num of Input: {self.num_of_input}, Num of Output: {self.num_of_output})' |
| 213 | |
| 214 | def __getstate__(self) -> dict: |
| 215 | state = super().__getstate__() |
| 216 | state['_input_vars'] = [var.name for var in self.inputs] |
no outgoing calls
no test coverage detected