(self, raw_text: str)
| 2472 | return mapping.get(str(aspect_ratio or "").strip(), "16:9") |
| 2473 | |
| 2474 | def _parse_sse_json_events(self, raw_text: str) -> List[Dict[str, Any]]: |
| 2475 | events: List[Dict[str, Any]] = [] |
| 2476 | if not raw_text: |
| 2477 | return events |
| 2478 | |
| 2479 | for block in raw_text.split("\n\n"): |
| 2480 | block = block.strip() |
| 2481 | if not block: |
| 2482 | continue |
| 2483 | |
| 2484 | data_lines = [line[5:].strip() for line in block.splitlines() if line.startswith("data:")] |
| 2485 | if not data_lines: |
| 2486 | continue |
| 2487 | |
| 2488 | payload_text = "\n".join(data_lines).strip() |
| 2489 | if not payload_text or payload_text == "[DONE]": |
| 2490 | continue |
| 2491 | |
| 2492 | try: |
| 2493 | payload = json.loads(payload_text) |
| 2494 | except Exception: |
| 2495 | continue |
| 2496 | |
| 2497 | if isinstance(payload, dict): |
| 2498 | events.append(payload) |
| 2499 | |
| 2500 | return events |
| 2501 | |
| 2502 | def _extract_agent_session_id(self, sessions_payload: Dict[str, Any]) -> Optional[str]: |
| 2503 | if not isinstance(sessions_payload, dict): |
no outgoing calls
no test coverage detected