52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
|
from typing import Dict, List, Any
|
||
|
|
from scapy.all import Packet
|
||
|
|
|
||
|
|
from .base import FrameTypeInterface
|
||
|
|
|
||
|
|
|
||
|
|
class Ch10TMATSStats(FrameTypeInterface):
|
||
|
|
"""Chapter 10 TMATS (Telemetry Attributes Transfer Standard) Statistics"""
|
||
|
|
def __init__(self):
|
||
|
|
super().__init__()
|
||
|
|
self.name = "Chapter 10 TMATS"
|
||
|
|
self.count = 0
|
||
|
|
self.bytes = 0
|
||
|
|
self.first_time = None
|
||
|
|
self.last_time = None
|
||
|
|
self.tmats_versions = set()
|
||
|
|
self.configuration_changes = 0
|
||
|
|
self.g_records = 0 # Data source records
|
||
|
|
self.r_records = 0 # Tape/storage records
|
||
|
|
self.m_records = 0 # Multiplexing/modulation records
|
||
|
|
|
||
|
|
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, parse TMATS records
|
||
|
|
|
||
|
|
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,
|
||
|
|
'Versions': len(self.tmats_versions),
|
||
|
|
'Config Changes': self.configuration_changes,
|
||
|
|
'G-Records': self.g_records,
|
||
|
|
'R-Records': self.r_records,
|
||
|
|
'M-Records': self.m_records
|
||
|
|
}
|
||
|
|
|
||
|
|
def get_column_definitions(self) -> List[tuple]:
|
||
|
|
return [
|
||
|
|
('Pkts', 'd'),
|
||
|
|
('Bytes', 'd'),
|
||
|
|
('Versions', 'd'),
|
||
|
|
('Config Changes', 'd'),
|
||
|
|
('G-Records', 'd'),
|
||
|
|
('R-Records', 'd'),
|
||
|
|
('M-Records', 'd')
|
||
|
|
]
|