Holds the scrolling window (time axis) of LatencyRecord objects.
| 283 | |
| 284 | |
| 285 | class ArrayOfLatencyRecords: |
| 286 | """ |
| 287 | Holds the scrolling window (time axis) of LatencyRecord objects. |
| 288 | """ |
| 289 | BLUE_PALETTE = {0: 15, 1: 51, 2: 45, 3: 39, 4: 33, 5: 27, 6: 21} # white→deep blue bg |
| 290 | RED_PALETTE = {0: 15, 1: 226, 2: 220, 3: 214, 4: 208, 5: 202, 6: 196} # white→red bg |
| 291 | ESC_RESET = "\x1b[0m" |
| 292 | |
| 293 | def __init__(self) -> None: |
| 294 | self.sample_number: int = 0 |
| 295 | # IMPORTANT: create distinct empty records (no shared reference) |
| 296 | self.data: List[LatencyRecord] = [LatencyRecord() for _ in range(0, g_params.num_latency_records + 1)] |
| 297 | |
| 298 | # ------------------------------- Debug -------------------------------- # |
| 299 | |
| 300 | def print_frequency_histograms_debug(self) -> None: |
| 301 | print('\nFrequency histograms:') |
| 302 | for record in self.data: |
| 303 | print(record.frequency_histogram, record.delta_time) |
| 304 | |
| 305 | def print_intensity_histograms_debug(self) -> None: |
| 306 | print('\nIntensity histograms:') |
| 307 | for record in self.data: |
| 308 | print(record.intensity_histogram, record.delta_time) |
| 309 | |
| 310 | # ------------------------------ Charting ------------------------------ # |
| 311 | |
| 312 | def add_new_record(self, record: LatencyRecord) -> None: |
| 313 | # Scroll window: discard oldest, append newest |
| 314 | self.data = self.data[1:] + [record] |
| 315 | self.sample_number += 1 |
| 316 | |
| 317 | @staticmethod |
| 318 | def _bg_color(token: int, palette: str) -> str: |
| 319 | if palette == 'blue': |
| 320 | c = ArrayOfLatencyRecords.BLUE_PALETTE[token] |
| 321 | elif palette == 'red': |
| 322 | c = ArrayOfLatencyRecords.RED_PALETTE[token] |
| 323 | else: |
| 324 | raise ValueError("palette must be 'blue' or 'red'") |
| 325 | return f"\x1b[48;5;{c}m" |
| 326 | |
| 327 | @staticmethod |
| 328 | def _fmt_value(v: float) -> str: |
| 329 | if v < 0: |
| 330 | return "0" |
| 331 | if v <= 9: |
| 332 | return f"{v:.1g}" |
| 333 | if v < 1_000_000: |
| 334 | return str(int(round(v))) |
| 335 | return f"{v:.2g}" |
| 336 | |
| 337 | @staticmethod |
| 338 | def _bucket_ms_label(bucket_exp: int) -> str: |
| 339 | """ |
| 340 | Render the bucket upper bound as milliseconds for the left axis. |
| 341 | - Always display in ms. |
| 342 | - For <1ms: show with leading dot, e.g., .512 |