Spaces:
Sleeping
Sleeping
| # 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) | |