in client/commands/profile.py [0:0]
def to_traceevents(events: Sequence[Event]) -> List[Dict[str, Any]]:
def to_traceevent(event: Event) -> Optional[Dict[str, Any]]:
if isinstance(event, DurationEvent):
duration_us = event.duration
start_time_us = event.metadata.timestamp - duration_us
return {
"pid": event.metadata.worker_id,
"tid": event.metadata.pid,
"ts": start_time_us,
"ph": "X",
"name": event.metadata.name,
"dur": duration_us,
"args": event.metadata.tags,
}
elif isinstance(event, CounterEvent):
timestamp_us = event.metadata.timestamp
arguments: Dict[str, Any] = {
key: int(value) for key, value in event.metadata.tags.items()
}
return {
"pid": event.metadata.worker_id,
"tid": event.metadata.pid,
"ts": timestamp_us,
"ph": "C",
"name": event.metadata.name,
"args": arguments,
}
else:
return None
return [
trace_event
for trace_event in map(to_traceevent, events)
if trace_event is not None
]