""" Bottom panel - Timeline visualization """ from typing import List, Tuple, Optional import curses from ...models import FlowStats, FrameTypeStats class TimelinePanel: """Bottom panel for timeline visualization""" def draw(self, stdscr, x_offset: int, y_offset: int, width: int, height: int, flows_list: List[FlowStats], selected_flow: int): """Draw timeline visualization panel for selected flow or frame type""" if not flows_list or height < 5: return # Get the selected flow and frame type flow, selected_frame_type = self._get_selected_flow_and_frame_type(flows_list, selected_flow) if not flow: return try: # Panel header stdscr.addstr(y_offset, x_offset, "TIMING VISUALIZATION", curses.A_BOLD) if selected_frame_type: stdscr.addstr(y_offset + 1, x_offset, f"Flow: {flow.src_ip} -> {flow.dst_ip} | Frame Type: {selected_frame_type}") else: stdscr.addstr(y_offset + 1, x_offset, f"Flow: {flow.src_ip} -> {flow.dst_ip} | All Frames") # Get the appropriate data for timeline if selected_frame_type and selected_frame_type in flow.frame_types: # Use frame type specific data ft_stats = flow.frame_types[selected_frame_type] if len(ft_stats.inter_arrival_times) < 2: stdscr.addstr(y_offset + 2, x_offset, f"Insufficient data for {selected_frame_type} timeline") return deviations = self._calculate_frame_type_deviations(ft_stats) timeline_flow = ft_stats # Use frame type stats for timeline else: # Use overall flow data if len(flow.inter_arrival_times) < 2: stdscr.addstr(y_offset + 2, x_offset, "Insufficient data for timeline") return deviations = self._calculate_frame_deviations(flow) timeline_flow = flow # Use overall flow stats for timeline if not deviations: stdscr.addstr(y_offset + 2, x_offset, "No timing data available") return # Timeline dimensions timeline_width = width - 10 # Leave space for labels timeline_height = height - 6 # Leave space for header, labels, and time scale timeline_y = y_offset + 3 timeline_x = x_offset + 5 # Draw timeline self._draw_ascii_timeline(stdscr, timeline_x, timeline_y, timeline_width, timeline_height, deviations, timeline_flow) except curses.error: # Ignore curses errors from writing outside screen bounds pass def _get_selected_flow_and_frame_type(self, flows_list: List[FlowStats], selected_flow: int) -> Tuple[Optional[FlowStats], Optional[str]]: """Get the currently selected flow and frame type based on selection index""" current_item = 0 for flow in flows_list: if current_item == selected_flow: return flow, None # Selected the main flow current_item += 1 # Check frame types for this flow if flow.frame_types: sorted_frame_types = sorted(flow.frame_types.items(), key=lambda x: x[1].count, reverse=True) for frame_type, ft_stats in sorted_frame_types: if current_item == selected_flow: return flow, frame_type # Selected a frame type current_item += 1 # Fallback to first flow if selection is out of bounds return flows_list[0] if flows_list else None, None def _calculate_frame_deviations(self, flow: FlowStats) -> List[Tuple[int, float]]: """Calculate frame deviations from average inter-arrival time""" if len(flow.inter_arrival_times) < 1 or flow.avg_inter_arrival == 0: return [] deviations = [] # Each inter_arrival_time[i] is between frame[i] and frame[i+1] for i, inter_time in enumerate(flow.inter_arrival_times): if i + 1 < len(flow.frame_numbers): frame_num = flow.frame_numbers[i + 1] # The frame that this inter-arrival time leads to deviation = inter_time - flow.avg_inter_arrival deviations.append((frame_num, deviation)) return deviations def _calculate_frame_type_deviations(self, ft_stats: FrameTypeStats) -> List[Tuple[int, float]]: """Calculate frame deviations for a specific frame type""" if len(ft_stats.inter_arrival_times) < 1 or ft_stats.avg_inter_arrival == 0: return [] deviations = [] # Each inter_arrival_time[i] is between frame[i] and frame[i+1] for i, inter_time in enumerate(ft_stats.inter_arrival_times): if i + 1 < len(ft_stats.frame_numbers): frame_num = ft_stats.frame_numbers[i + 1] # The frame that this inter-arrival time leads to deviation = inter_time - ft_stats.avg_inter_arrival deviations.append((frame_num, deviation)) return deviations def _draw_ascii_timeline(self, stdscr, x_offset: int, y_offset: int, width: int, height: int, deviations: List[Tuple[int, float]], flow): """Draw ASCII timeline chart""" if not deviations or width < 10 or height < 3: return # Find min/max deviations for scaling deviation_values = [dev for _, dev in deviations] max_deviation = max(abs(min(deviation_values)), max(deviation_values)) if max_deviation == 0: max_deviation = 0.001 # Avoid division by zero # Calculate center line center_y = y_offset + height // 2 # Draw center line (represents average timing) center_line = "─" * width stdscr.addstr(center_y, x_offset, center_line) # Add center line label if x_offset > 4: stdscr.addstr(center_y, x_offset - 4, "AVG") # Scale factor for vertical positioning scale_factor = (height // 2) / max_deviation # Always scale to use the entire width # Calculate the time span of the data if len(flow.timestamps) < 2: return start_time = flow.timestamps[0] end_time = flow.timestamps[-1] time_span = end_time - start_time if time_span <= 0: return # Create a mapping from deviation frame numbers to actual timestamps frame_to_timestamp = {} for i, (frame_num, deviation) in enumerate(deviations): if i < len(flow.timestamps): frame_to_timestamp[frame_num] = flow.timestamps[i] # Plot points across entire width for x in range(width): # Calculate which timestamp this x position represents time_ratio = x / (width - 1) if width > 1 else 0 target_time = start_time + (time_ratio * time_span) # Find the closest deviation to this time closest_deviation = None min_time_diff = float('inf') for frame_num, deviation in deviations: # Use the correct timestamp mapping if frame_num in frame_to_timestamp: frame_time = frame_to_timestamp[frame_num] time_diff = abs(frame_time - target_time) if time_diff < min_time_diff: min_time_diff = time_diff closest_deviation = deviation if closest_deviation is not None: # Calculate vertical position y_pos = center_y - int(closest_deviation * scale_factor) y_pos = max(y_offset, min(y_offset + height - 1, y_pos)) # Choose character based on deviation magnitude char = self._get_timeline_char(closest_deviation, flow.avg_inter_arrival) # Draw the point try: stdscr.addstr(y_pos, x_offset + x, char) except curses.error: pass # Draw scale labels and timeline info self._draw_timeline_labels(stdscr, x_offset, y_offset, width, height, max_deviation, deviations, flow, time_span) def _get_timeline_char(self, deviation: float, avg_time: float) -> str: """Get character representation for timeline point based on deviation""" if abs(deviation) < avg_time * 0.1: # Within 10% of average return "·" elif abs(deviation) < avg_time * 0.5: # Within 50% of average return "•" if deviation > 0 else "○" else: # Significant deviation (outlier) return "█" if deviation > 0 else "▄" def _draw_timeline_labels(self, stdscr, x_offset: int, y_offset: int, width: int, height: int, max_deviation: float, deviations: List[Tuple[int, float]], flow, time_span: float): """Draw timeline labels and summary information""" # Draw scale labels if height >= 5: # Top label (positive deviation) top_dev = max_deviation if x_offset > 4: stdscr.addstr(y_offset, x_offset - 4, f"+{top_dev:.2f}s") # Bottom label (negative deviation) bottom_dev = -max_deviation if x_offset > 4: stdscr.addstr(y_offset + height - 1, x_offset - 4, f"{bottom_dev:.2f}s") # Timeline info with time scale above summary info_y = y_offset + height + 1 if info_y < y_offset + height + 3: # Make sure we have space for two lines total_frames = len(deviations) # First line: Time scale relative_start = 0.0 relative_end = time_span relative_middle = time_span / 2 # Format time scale labels start_label = f"{relative_start:.1f}s" middle_label = f"{relative_middle:.1f}s" end_label = f"{relative_end:.1f}s" # Draw time scale labels at left, middle, right stdscr.addstr(info_y, x_offset, start_label) # Middle label middle_x = x_offset + width // 2 - len(middle_label) // 2 if middle_x > x_offset + len(start_label) + 1 and middle_x + len(middle_label) < x_offset + width - len(end_label) - 1: stdscr.addstr(info_y, middle_x, middle_label) # Right label end_x = x_offset + width - len(end_label) if end_x > x_offset + len(start_label) + 1: stdscr.addstr(info_y, end_x, end_label) # Second line: Frame count and deviation range summary_y = info_y + 1 if summary_y < y_offset + height + 3: left_info = f"Frames: {total_frames} | Range: ±{max_deviation:.3f}s" stdscr.addstr(summary_y, x_offset, left_info) # Right side outliers count with 2σ threshold threshold_2sigma = flow.avg_inter_arrival + (2 * flow.std_inter_arrival) outliers_info = f"Outliers: {len(flow.outlier_frames)} (>2σ: {threshold_2sigma:.4f}s)" outliers_x = x_offset + width - len(outliers_info) if outliers_x > x_offset + len(left_info) + 2: # Make sure there's space stdscr.addstr(summary_y, outliers_x, outliers_info)