Spaces:
Sleeping
Sleeping
File size: 3,805 Bytes
936432e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
# call_flow.py
def create_call_flow_diagram(calls_by_id):
"""
Given a dictionary of Call objects (keyed by call_id), generate
a textual call flow for each call. We assume each Call has a
'sip_sequence' attribute: a list of dicts with:
{
'timestamp': datetime object,
'src_ip': str,
'dst_ip': str,
'message': str # e.g. "INVITE (Call-ID: abc)"
}
The output will look like a classic ASCII SIP ladder diagram.
Returns a string that concatenates the diagrams for all calls.
"""
if not calls_by_id:
return "No calls found to display."
# We'll combine diagrams for each call into one text output
output = []
for call_id, call_obj in calls_by_id.items():
# Retrieve the SIP sequence in chronological order (by timestamp)
# You must ensure your parser sets call_obj.sip_sequence sorted by time
sip_sequence = getattr(call_obj, 'sip_sequence', [])
sip_sequence = sorted(sip_sequence, key=lambda x: x['timestamp'])
# Get unique participants for a horizontal layout
# We'll just gather distinct IP addresses from the sequence
# and put them left (caller) -> right (callee).
# If you know which is the caller vs. callee, you can fix that order.
participants = sorted(
list({msg['src_ip'] for msg in sip_sequence} | {msg['dst_ip'] for msg in sip_sequence})
)
if len(participants) < 2:
# If we somehow only have one or zero participants, skip
output.append(f"Call-ID {call_id} only has one participant.\n")
continue
# We’ll just place the first IP as “Left” and last IP as “Right” for demonstration
left_participant = participants[0]
right_participant = participants[-1]
# Header for this call
diagram_lines = []
diagram_lines.append(f"Call Flow for Call-ID: {call_id}")
diagram_lines.append(f" {left_participant:<30} {right_participant}")
diagram_lines.append(" -----------------------------------------------------------")
# Each SIP message in chronological order
for msg in sip_sequence:
src = msg['src_ip']
dst = msg['dst_ip']
message = msg['message']
if src == left_participant and dst == right_participant:
# Left -> Right
diagram_lines.append(f" {message:<30} ----------------->")
elif src == right_participant and dst == left_participant:
# Right -> Left
# We want the message on the right side, but in ASCII, we can do:
# Some spaces, then the message, then <----
# For demonstration, we do something simpler:
diagram_lines.append(f" {message}")
diagram_lines.append(" <-------------------------------")
else:
# If the message is between some other IP pair, we can either
# skip or try to align it in the middle. For now, just note it.
# This might happen if there are multiple proxies or servers.
diagram_lines.append(f" [{src} -> {dst}] {message}")
# Example: If we want to artificially place an "RTP ...." line
# you could do a simple check if the call has media streams:
if call_obj.media_streams:
diagram_lines.append(" RTP ...................... RTP ....................")
# Add a blank line after the call’s diagram
output.append("\n".join(diagram_lines) + "\n")
return "\n".join(output)
|