File indexing completed on 2026-11-09 10:47:33
0001 """Measured Event Service phases, with aggregate legacy time kept explicit."""
0002 from datetime import datetime, timezone
0003
0004
0005 def epoch(value):
0006 if value is None:
0007 return None
0008 if isinstance(value, (int, float)):
0009 return float(value)
0010 parsed = datetime.fromisoformat(str(value).replace('Z', '+00:00'))
0011 if parsed.tzinfo is None:
0012 parsed = parsed.replace(tzinfo=timezone.utc)
0013 return parsed.timestamp()
0014
0015
0016 def pooled(unit):
0017 """True when the unit's reconstruction ran on a shared pool member
0018 (swf-epicprod es/reco_pool.py), not on the unit's own core."""
0019 return (unit.get('reconstruction') or {}).get('pool_member') is not None
0020
0021
0022 def member_segments(unit, origin, observed_end):
0023 """A pooled unit's reconstruction on its member's lane: the member's
0024 startup when this unit's request started it, then the request."""
0025 reconstruction = unit.get('reconstruction') or {}
0026 parts = []
0027 for phase, first, last in (('initialization', reconstruction.get('initialization_started_at'),
0028 reconstruction.get('initialized_at')),
0029 ('reconstruction', reconstruction.get('started_at'),
0030 reconstruction.get('ended_at'))):
0031 first, last = epoch(first), epoch(last)
0032 if phase == 'initialization' and int(reconstruction.get('pool_requests') or 1) != 1:
0033 continue
0034 if first is None or last is None:
0035 continue
0036 first, last = max(origin, first), min(observed_end, last)
0037 if last > first:
0038 parts.append({'phase': phase, 'start_s': first - origin, 'end_s': last - origin})
0039 return parts
0040
0041
0042 def unit_segments(unit, origin, observed_end, without_reconstruction=False):
0043 """Partition one unit without treating an unmeasured interval as processing.
0044 without_reconstruction: the reconstruction ran on a shared member, so it is
0045 no part of this core's time (member_segments draws it on the member's lane)."""
0046 start = max(origin, epoch(unit['started_at']))
0047 end = min(observed_end, epoch(unit.get('ended_at')) or observed_end)
0048 if end <= start:
0049 return []
0050 candidates = []
0051
0052 def add(phase, first, last, priority):
0053 first, last = epoch(first), epoch(last)
0054 if first is None or last is None:
0055 return
0056 first, last = max(start, first), min(end, last)
0057 if last > first:
0058 candidates.append((first, last, phase, priority))
0059
0060 simulation = unit.get('simulation') or {}
0061 reconstruction = unit.get('reconstruction') or {}
0062 phases = unit.get('phases') or {}
0063
0064 if simulation.get('initialization_started_at') and int(simulation.get('sequence') or 1) == 1:
0065 add('initialization', simulation['initialization_started_at'],
0066 simulation.get('first_event_at') or end, 3)
0067 if simulation.get('first_event_at'):
0068 add('simulation', simulation['first_event_at'],
0069 simulation.get('last_event_at') or simulation.get('ended_at') or end, 2)
0070 if reconstruction.get('initialization_started_at') and not without_reconstruction:
0071 add('initialization', reconstruction['initialization_started_at'],
0072 reconstruction.get('initialized_at') or end, 3)
0073 if reconstruction.get('started_at') and not without_reconstruction:
0074 add('reconstruction', reconstruction['started_at'], reconstruction.get('ended_at') or end, 2)
0075 for name in ('simulation', 'reconstruction'):
0076 phase = phases.get(name) or {}
0077 separated = (bool(simulation.get('first_event_at')) if name == 'simulation'
0078 else bool(reconstruction.get('started_at')))
0079 if name == 'reconstruction' and without_reconstruction:
0080 continue
0081 if phase.get('started_at') and not separated:
0082 add('unseparated', phase['started_at'], phase.get('ended_at') or end, 1)
0083 measured = bool(simulation or reconstruction or phases)
0084 add('overhead' if measured else 'unseparated', start, end, 0)
0085 boundaries = sorted({point for a, b, _, _ in candidates for point in (a, b)})
0086 result = []
0087 for first, last in zip(boundaries, boundaries[1:]):
0088 middle = (first + last) / 2
0089 covering = [part for part in candidates if part[0] <= middle < part[1]]
0090 if not covering:
0091 continue
0092 phase = max(covering, key=lambda part: part[3])[2]
0093 if result and result[-1]['phase'] == phase:
0094 result[-1]['end_s'] = last - origin
0095 else:
0096 result.append({'phase': phase, 'start_s': first - origin, 'end_s': last - origin})
0097 return result
0098
0099
0100 def turnaround(lanes):
0101 """On each core, the seconds from the last measured event processing of
0102 one unit to the first of the next: what a core spends between segments.
0103 A unit without measured processing breaks the sequence rather than
0104 joining two gaps into one."""
0105 gaps = []
0106 for units in lanes:
0107 previous = None
0108 for unit in sorted(units, key=lambda u: u['start_s']):
0109 processing = [part for part in unit.get('segments') or []
0110 if part['phase'] in ('simulation', 'reconstruction')]
0111 if not processing:
0112 previous = None
0113 continue
0114 if previous is not None:
0115 gaps.append(max(0., processing[0]['start_s'] - previous))
0116 previous = processing[-1]['end_s']
0117 if not gaps:
0118 return {'turnaround_n': 0}
0119 gaps.sort()
0120
0121 def quantile(q):
0122 return round(gaps[min(len(gaps) - 1, int(q * len(gaps)))], 2)
0123
0124 return {'turnaround_n': len(gaps), 'turnaround_median_s': quantile(0.5),
0125 'turnaround_p90_s': quantile(0.9), 'turnaround_max_s': round(gaps[-1], 2)}
0126
0127
0128 def phase_totals(units):
0129 result = dict(processing_s=0., saved_processing_s=0., initialization_s=0.,
0130 overhead_s=0., unseparated_s=0., measured_units=0)
0131 for unit in units:
0132 separated = False
0133 for part in unit.get('segments') or []:
0134 duration = max(0., part['end_s'] - part['start_s'])
0135 phase = part['phase']
0136 if phase in ('simulation', 'reconstruction'):
0137 separated = True
0138 result['processing_s'] += duration
0139 if unit['status'] == 'done':
0140 result['saved_processing_s'] += duration
0141 elif phase == 'initialization':
0142 separated = True
0143 result['initialization_s'] += duration
0144 else:
0145 result[phase + '_s'] += duration
0146 result['measured_units'] += int(separated)
0147 return {key: round(value, 3) for key, value in result.items()}
