(self, request, context)
| 134 | return backend_pb2.VoiceEmbedResponse(embedding=list(vec), model=self.model_name) |
| 135 | |
| 136 | def VoiceAnalyze(self, request, context): |
| 137 | engine = self._require_engine(context) |
| 138 | if engine is None: |
| 139 | return backend_pb2.VoiceAnalyzeResponse() |
| 140 | if not request.audio: |
| 141 | context.set_code(grpc.StatusCode.INVALID_ARGUMENT) |
| 142 | context.set_details("audio is required") |
| 143 | return backend_pb2.VoiceAnalyzeResponse() |
| 144 | |
| 145 | actions = list(request.actions) or ["age", "gender", "emotion"] |
| 146 | try: |
| 147 | segments = engine.analyze(request.audio, actions) |
| 148 | except NotImplementedError: |
| 149 | context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
| 150 | context.set_details(f"analyze not supported by {self.engine_name}") |
| 151 | return backend_pb2.VoiceAnalyzeResponse() |
| 152 | except Exception as exc: # noqa: BLE001 |
| 153 | context.set_code(grpc.StatusCode.INTERNAL) |
| 154 | context.set_details(f"voice analyze failed: {exc}") |
| 155 | return backend_pb2.VoiceAnalyzeResponse() |
| 156 | |
| 157 | proto_segments = [] |
| 158 | for seg in segments: |
| 159 | proto_segments.append( |
| 160 | backend_pb2.VoiceAnalysis( |
| 161 | start=seg.get("start", 0.0), |
| 162 | end=seg.get("end", 0.0), |
| 163 | age=seg.get("age", 0.0), |
| 164 | dominant_gender=seg.get("dominant_gender", ""), |
| 165 | gender=seg.get("gender", {}), |
| 166 | dominant_emotion=seg.get("dominant_emotion", ""), |
| 167 | emotion=seg.get("emotion", {}), |
| 168 | ) |
| 169 | ) |
| 170 | return backend_pb2.VoiceAnalyzeResponse(segments=proto_segments) |
| 171 | |
| 172 | |
| 173 | def serve(address: str) -> None: |
nothing calls this directly
no test coverage detected