(self, request, context)
| 718 | return backend_pb2.TranscriptResult() |
| 719 | |
| 720 | async def AudioTranscriptionStream(self, request, context): |
| 721 | if self.whisper_model is None or self.whisper_tokenizer is None: |
| 722 | context.set_code(grpc.StatusCode.FAILED_PRECONDITION) |
| 723 | context.set_details("No Whisper model loaded") |
| 724 | return |
| 725 | |
| 726 | try: |
| 727 | if not request.dst: |
| 728 | context.set_code(grpc.StatusCode.INVALID_ARGUMENT) |
| 729 | context.set_details("TranscriptRequest.dst (audio file path) is required") |
| 730 | return |
| 731 | |
| 732 | # The vendored tinygrad whisper loop is chunked at the file level |
| 733 | # (one inference pass per 30s segment), not token-level. To still |
| 734 | # produce a streaming response we run the full transcription and |
| 735 | # emit it as a single delta + a final-result envelope so the client |
| 736 | # gets both code paths exercised. |
| 737 | text, duration = self._transcribe(request.dst, request.language) |
| 738 | yield backend_pb2.TranscriptStreamResponse(delta=text) |
| 739 | final = backend_pb2.TranscriptResult( |
| 740 | text=text, |
| 741 | language=request.language or "en", |
| 742 | duration=duration, |
| 743 | segments=[backend_pb2.TranscriptSegment(id=0, start=0, end=0, text=text)], |
| 744 | ) |
| 745 | yield backend_pb2.TranscriptStreamResponse(final_result=final) |
| 746 | except Exception as exc: |
| 747 | import traceback |
| 748 | traceback.print_exc() |
| 749 | context.set_code(grpc.StatusCode.INTERNAL) |
| 750 | context.set_details(f"AudioTranscriptionStream failed: {exc}") |
| 751 | |
| 752 | async def Status(self, request, context): |
| 753 | return backend_pb2.StatusResponse(state=backend_pb2.StatusResponse.READY) |
nothing calls this directly
no test coverage detected