| 37 | |
| 38 | |
| 39 | class ScanStats: |
| 40 | def __init__(self, scan): |
| 41 | self.scan = scan |
| 42 | self.module_stats = {} |
| 43 | self.events_emitted_by_type = {} |
| 44 | self.speedometer = SpeedCounter(scan.status_frequency) |
| 45 | |
| 46 | def event_produced(self, event): |
| 47 | _increment(self.events_emitted_by_type, event.type) |
| 48 | module_stat = self.get(event.module) |
| 49 | if module_stat is not None: |
| 50 | module_stat.increment_produced(event) |
| 51 | |
| 52 | def event_consumed(self, event, module): |
| 53 | self.speedometer.tick() |
| 54 | # skip ingress/egress modules, etc. |
| 55 | if module.name.startswith("_"): |
| 56 | return |
| 57 | module_stat = self.get(module) |
| 58 | if module_stat is not None: |
| 59 | module_stat.increment_consumed(event) |
| 60 | |
| 61 | def get(self, module): |
| 62 | try: |
| 63 | module_stat = self.module_stats[module.name] |
| 64 | except KeyError: |
| 65 | module_stat = ModuleStat(module) |
| 66 | self.module_stats[module.name] = module_stat |
| 67 | except AttributeError: |
| 68 | module_stat = None |
| 69 | return module_stat |
| 70 | |
| 71 | def table(self): |
| 72 | header = ["Module", "Produced", "Consumed"] |
| 73 | table = [] |
| 74 | for mname, mstat in self.module_stats.items(): |
| 75 | if mname == "TARGET" or mstat.module._stats_exclude: |
| 76 | continue |
| 77 | table_row = [] |
| 78 | table_row.append(mname) |
| 79 | produced_str = f"{mstat.produced_total:,}" |
| 80 | produced = sorted(mstat.produced.items(), key=lambda x: x[0]) |
| 81 | if produced: |
| 82 | produced_str += " (" + ", ".join(f"{c:,} {t}" for t, c in produced) + ")" |
| 83 | table_row.append(produced_str) |
| 84 | consumed_str = f"{mstat.consumed_total:,}" |
| 85 | consumed = sorted(mstat.consumed.items(), key=lambda x: x[0]) |
| 86 | if consumed: |
| 87 | consumed_str += " (" + ", ".join(f"{c:,} {t}" for t, c in consumed) + ")" |
| 88 | table_row.append(consumed_str) |
| 89 | table.append(table_row) |
| 90 | table.sort(key=lambda x: self.module_stats[x[0]].produced_total, reverse=True) |
| 91 | return [header] + table |
| 92 | |
| 93 | def _make_table(self): |
| 94 | table = self.table() |
| 95 | if len(table) == 1: |
| 96 | table += [["None", "None", "None"]] |
no outgoing calls
no test coverage detected
searching dependent graphs…