(self, *pipes)
| 201 | print("Interrupted by user.") |
| 202 | |
| 203 | def add(self, *pipes): |
| 204 | # type: (*Pipe) -> None |
| 205 | _pipes = self._add_pipes(*pipes) |
| 206 | with self.command_lock: |
| 207 | if self.thread is not None: |
| 208 | for p in _pipes: |
| 209 | p.start() |
| 210 | self._write_cmd("A") |
| 211 | |
| 212 | def graph(self, **kargs): |
| 213 | # type: (Any) -> None |
no test coverage detected