from collections import defaultdict from typing import Dict, List, Any from scapy.all import Packet from .base import FrameTypeInterface class Ch10AnalogStats(FrameTypeInterface): """Chapter 10 Analog Data Statistics""" def __init__(self): super().__init__() self.name = "Chapter 10 Analog" self.count = 0 self.bytes = 0 self.first_time = None self.last_time = None self.channel_ids = set() self.sample_counts = defaultdict(int) self.data_gaps = 0 self.overrange_count = 0 self.underrange_count = 0 def add(self, timestamp: float, size: int, packet: Packet): self.count += 1 self.bytes += size if self.first_time is None: self.first_time = timestamp self.last_time = timestamp # In real implementation, decode Ch10 analog headers # Extract channel ID, sample count, check for over/underrange def get_summary_dict(self) -> Dict[str, Any]: duration = (self.last_time or 0) - (self.first_time or 0) return { 'Pkts': self.count, 'Bytes': self.bytes, 'Duration': round(duration, 3), 'Channels': len(self.channel_ids), 'Data Gaps': self.data_gaps, 'Overrange': self.overrange_count, 'Underrange': self.underrange_count, 'Pkt/s': round(self.count / duration, 1) if duration > 0 else 0 } def get_column_definitions(self) -> List[tuple]: return [ ('Pkts', 'd'), ('Bytes', 'd'), ('Duration', '.3f'), ('Channels', 'd'), ('Data Gaps', 'd'), ('Overrange', 'd'), ('Underrange', 'd'), ('Pkt/s', '.1f') ]