openStream method

Future<SttStream> openStream(AudioFormatSpec format, { SttOptions? options, })

Open a live transcription stream. The format is established once; every frame pushed afterward carries raw PCM in that format.

final stream = await RunAnywhere.stt.openStream(
  const AudioFormatSpec(encoding: AudioEncoding.pcm16, sampleRate: 16000));
stream.events.listen(print);

Throws SDKException when format is a container encoding (live streams take raw PCM only), no STT model is loaded, or the native streaming session cannot start.

Implementation

Future<SttStream> openStream(
  AudioFormatSpec format, {
  SttOptions? options,
}) async {
  if (format.encoding == AudioEncoding.container) {
    throw SDKException.invalidInput(
      'stt.openStream needs raw PCM (pcm16/float32), not a container format',
    );
  }
  if (!DartBridge.isInitialized) {
    throw SDKException.notInitialized();
  }
  await DartBridge.ensureServicesReady();
  final current = await RunAnywhereModelLifecycle.shared.current(
    model_pb.CurrentModelRequest(category: _category),
  );
  if (!current.found) {
    throw SDKException.componentNotReady('STT');
  }
  final modelId = current.modelId.isNotEmpty
      ? current.modelId
      : current.model.id;
  final modelPath = current.resolvedPath.isNotEmpty
      ? current.resolvedPath
      : current.model.localPath;
  if (modelId.isEmpty || modelPath.isEmpty) {
    throw SDKException.modelLoadFailed(
      modelId,
      'Loaded STT model is missing a resolved path',
    );
  }
  DartBridgeSTT.shared.loadModelForStreaming(
    path: modelPath,
    id: modelId,
    name: current.model.name.isNotEmpty ? current.model.name : modelId,
  );

  final requestId = 'stt-${DateTime.now().microsecondsSinceEpoch}';
  final frameController = StreamController<Uint8List>();
  final eventController = StreamController<TranscriptionEvent>();
  var announcedStarted = false;
  var finished = false;
  var closed = false;

  void announceStarted() {
    if (announcedStarted) return;
    announcedStarted = true;
    if (!eventController.isClosed) {
      eventController.add(TranscriptionStarted(requestId));
    }
  }

  Future<void> runSession() async {
    var sawTerminal = false;
    try {
      final partials = DartBridgeSTT.shared.transcribeSessionStream(
        frameController.stream,
        (options ?? SttOptions()).toProto(),
      );
      await for (final partial in partials) {
        if (eventController.isClosed) return;
        if (partial.isFinal) {
          sawTerminal = true;
          // `STTPartialResult.final_output` (a full `STTOutput`) was
          // deleted outright, along with `segment_index`/`confidence`/
          // `stability`/etc (idl/stt_options.proto) — `text` is the only
          // content field left on a final partial.
          eventController.add(
            TranscriptionFinal(Transcription(text: partial.text)),
          );
          eventController.add(const TranscriptionCompleted());
          break;
        }
        eventController.add(TranscriptionPartial(partial.text));
      }
      // Never fabricate a settled transcript: a session that ends
      // without a final result reports it honestly instead of
      // synthesizing an empty `TranscriptionFinal`.
      if (!sawTerminal && !eventController.isClosed) {
        eventController.add(
          TranscriptionFailed(
            SDKException.processingFailed(
              'Transcription stream ended before a final result',
            ),
          ),
        );
      }
    } catch (e) {
      if (!eventController.isClosed) {
        eventController.add(
          TranscriptionFailed(
            e is SDKException ? e : SDKException.processingFailed('$e'),
          ),
        );
      }
    } finally {
      unawaited(eventController.close());
    }
  }

  unawaited(runSession());

  return SttStream(
    events: eventController.stream,
    pushHandler: (frame) {
      if (closed || finished) return;
      announceStarted();
      if (!frameController.isClosed) frameController.add(frame.samples);
    },
    flushHandler: () {
      // Frames are fed to the native session as they arrive.
    },
    finishHandler: () {
      if (finished || closed) return;
      finished = true;
      announceStarted();
      unawaited(frameController.close());
    },
    closeHandler: () async {
      if (closed) return;
      closed = true;
      unawaited(frameController.close());
      await eventController.close();
    },
  );
}