Clear all statistics and reset state
(self)
| 1685 | if elapsed > 0: |
| 1686 | packets_delta = self.total_packets - self._last_packet_count |
| 1687 | bytes_delta = self.total_bytes - self._last_byte_count |
| 1688 | |
| 1689 | self.packets_per_second = packets_delta / elapsed |
| 1690 | self.bytes_per_second = bytes_delta / elapsed |
| 1691 | |
| 1692 | self._last_packet_count = self.total_packets |
| 1693 | self._last_byte_count = self.total_bytes |
| 1694 | self._last_metrics_time = current_time |
| 1695 | |
| 1696 | def on_alert(self, callback: Callable): |
| 1697 | """Register a callback for alerts""" |
| 1698 | self._on_alert_callbacks.append(callback) |
| 1699 | |
| 1700 | def _format_bytes(self, bytes_val: int) -> str: |
| 1701 | """Convert bytes to human readable format""" |
| 1702 | for unit in ['B', 'KB', 'MB', 'GB', 'TB']: |
| 1703 | if bytes_val < 1024: |
| 1704 | return f"{bytes_val:.1f} {unit}" |
| 1705 | bytes_val /= 1024 |
| 1706 | return f"{bytes_val:.1f} PB" |
| 1707 |