Publish each sample as a CSV row via UDP.
(self, data: np.ndarray)
| 166 | self.bytes_sent = 0 |
| 167 | |
| 168 | def publish(self, data: np.ndarray) -> None: |
| 169 | """Publish each sample as a CSV row via UDP.""" |
| 170 | num_channels, num_samples = data.shape |
| 171 | for i in range(num_samples): |
| 172 | row = ( |
| 173 | ",".join(self._fmt.format(data[ch, i]) for ch in range(num_channels)) |
| 174 | + "\n" |
| 175 | ) |
| 176 | payload = row.encode("utf-8") |
| 177 | try: |
| 178 | self.socket.sendto(payload, self._address) |
| 179 | self.packets_sent += 1 |
| 180 | self.bytes_sent += len(payload) |
| 181 | except socket.error: |
| 182 | pass |
| 183 | |
| 184 | def close(self) -> None: |
| 185 | self.socket.close() |
no test coverage detected