Project one streamed Envelope onto a typed frame.
(env: pb.Envelope)
| 114 | |
| 115 | |
| 116 | def _frame_from_envelope(env: pb.Envelope) -> AskFrame: |
| 117 | """Project one streamed Envelope onto a typed frame.""" |
| 118 | kind = env.kind |
| 119 | if kind == pb.Envelope.KIND_DONE: |
| 120 | d = env.done |
| 121 | return DoneFrame( |
| 122 | type="done", |
| 123 | success=d.success, |
| 124 | text=d.text or "", |
| 125 | error=d.error or "", |
| 126 | duration_ms=int(d.duration_ms or 0), |
| 127 | ) |
| 128 | if kind == pb.Envelope.KIND_CALLBACK: |
| 129 | arm = env.WhichOneof("payload") |
| 130 | if arm == "pin_required": |
| 131 | pr = env.pin_required |
| 132 | return PinRequiredFrame( |
| 133 | type="pin_required", challenge_id=pr.challenge_id, label=pr.label |
| 134 | ) |
| 135 | if arm == "confirm_required": |
| 136 | cr = env.confirm_required |
| 137 | return ConfirmRequiredFrame( |
| 138 | type="confirm_required", |
| 139 | token=cr.token, |
| 140 | plan=cr.plan, |
| 141 | danger_level=cr.danger_level or "medium", |
| 142 | ) |
| 143 | return UnknownFrame(type=str(arm)) |
| 144 | if kind == pb.Envelope.KIND_EVENT: |
| 145 | frame = env.ask_frame |
| 146 | inner = frame.WhichOneof("frame") |
| 147 | if inner == "event": |
| 148 | ev = frame.event |
| 149 | payload: Any = None |
| 150 | if ev.payload_json: |
| 151 | try: |
| 152 | payload = json.loads(ev.payload_json) |
| 153 | except (ValueError, json.JSONDecodeError): |
| 154 | payload = ev.payload_json |
| 155 | return EventFrame(type="event", event_type=int(ev.event_type), payload=payload) |
| 156 | if inner == "pin_denied": |
| 157 | pd = frame.pin_denied |
| 158 | return PinDeniedFrame( |
| 159 | type="pin_denied", challenge_id=pd.challenge_id, reason=pd.reason |
| 160 | ) |
| 161 | return UnknownFrame(type=str(inner)) |
| 162 | return UnknownFrame(type=pb.Envelope.Kind.Name(kind)) |
| 163 | |
| 164 | |
| 165 | # --- FrameStream ---------------------------------------------------------- |
no test coverage detected