Piping capured audio to insertable stream from shell script

[guest271314]

For the case of Chromium refusal to support capture of monitor devices am using Native Messaging and Native File System to write and read a file which is then parsed and set as outputs at AudioWorkletProcessor.process(), in pertinent part

parec --raw -d alsa_output.pci-0000_00_1b.0.analog-stereo.monitor ../app/output

which is read in main thread at browser. However, one issue is that Native File System currently cannot got a single handle on a file that is simultaneously being written to for the purpose of reading and writing at the same time, DOMExceptions will be thrown, and requires reading the entire file at each iteration to slice() from previous offset

          async function* fileStream() {
            while (true) {
              let fileHandle, fileBit, buffer;
                // if exception not thrown slice file from readOffset, handle exceptions
                // https://bugs.chromium.org/p/chromium/issues/detail?id=1084880
                // TODO: stream file being written at local filesystem
                // without reading entire file at each iteration before slice
                fileHandle = await dir.getFileHandle('output', {
                  create: false,
                });
                fileBit = await fileHandle.getFile();
                if (fileBit) {
                  const slice = fileBit.slice(readOffset);
                  if (slice.size === 0 && done) {
                    break;
                  }
                  if (slice.size > 0) {
                    buffer = await slice.arrayBuffer();
                    readOffset = readOffset + slice.size;
                    const u8_sab_view = new Uint8Array(memory.buffer);
                    const u8_file_view = new Uint8Array(buffer);
                    u8_sab_view.set(u8_file_view, writeOffset);
                    // accumulate  512 * 346 * 2 of data
                    if (
                      writeOffset > 512 * 346 * 2 &&
                      ac.state === 'suspended'
                    ) {
                      await ac.resume();
                    }
                    writeOffset = readOffset;
                  }
                }
              } catch (err) {
                // handle DOMException
                // : A requested file or directory could not be found at the time an operation was processed.
                // : The requested file could not be read, typically due to permission problems that have occurred after a reference to a file was acquired.
                if (
                  err instanceof DOMException ||
                  err instanceof TypeError ||
                  err
                ) {
                  console.warn(err);
                }
              } finally {
                yield;
              }
            }
          }
          for await (const _ of fileStream()) {
            if (done) break;
          }

Does opus-tools have the capability to create an Opus bitstream that will support piping the output therefrom to the writable side of the insertable stream? That is, instead of writing the file and reading the file we can do something like

parec --raw -d alsa_output.pci-0000_00_1b.0.analog-stereo.monitor | opusenc - - | opusenc <options_to_make_stdout_insertable_stream_writable_input> -

where we can then write() the output from the native shell script directly to an insertable stream, avoiding the need to re-read the same file just to get the current offset, or use SharedArrayBuffer to store the contents of the file in memory; we do not need to write a file at all, rather actually stream output from native application to RTCPeerConnection?

[guest271314]

Since WebRTC supports PCM how can we pipe directly from STDOUT (s16le 2ch 44100Hz) to insertable stream (writable)?

[guest271314]

One way to solve this is

fetch('url').then(r => r.mediaStream()) and fetch('url').then(r => r.mediaStreamTrack())

Proof-of-concept

<?php 
  if (isset($_GET["start"])) {
    header("Access-Control-Allow-Origin: *");
    header("Content-Type: application/octet-stream");
    echo passthru("parec -v --raw -d alsa_output.pci-0000_00_1b.0.analog-stereo.monitor");
    exit();
  }

<doctype html>
  <html>
    <body>
      <button>start</button><button>stop</button>
      <script>
        if (globalThis.gc) {
        //   gc();
        }

        let controller, signal;
        // throws memory allocation error, use multiple Memory instances
        // const memory = new WebAssembly.Memory({initial: 8062, maximum: 9675, shared: true});
        const [start, stop] = document.querySelectorAll('button');
        start.onclick = async e => {
          class AudioWorkletProcessor {}
          class AudioWorkletNativeFileStream extends AudioWorkletProcessor {
            constructor(options) {
              super(options);
              this.byteSize = 512 * 344 * 60 * 50;
              this.memory = new Uint8Array(this.byteSize);
              Object.assign(this, options.processorOptions);
              this.port.onmessage = this.appendBuffers.bind(this);
            }
            async appendBuffers({ data: readable }) {
              Object.assign(this, { readable });
              const source = {
                write: (value, c) => {
                  if (this.totalBytes < this.byteSize) {
                    this.memory.set(value, this.readOffset);
                    this.readOffset = this.readOffset + value.buffer.byteLength;
                    this.totalBytes = this.readOffset;
                  } else {
                    console.log(
                      value.buffer.byteLength,
                      this.readOffset,
                      this.totalBytes
                    );
                  }
                  if (this.totalBytes >= (384 * 512) / 2 && !this.started) {
                    this.started = true;
                    this.port.postMessage({ start: this.started });
                  }
                },
                close() {
                  console.log('stopped');
                },
              };
              try {
                await this.readable.pipeTo(new WritableStream(source));
              } catch (e) {
                console.warn(e);
                console.log(this.writeOffset, this.totalBytes);
                this.endOfStream();
              }
            }
            endOfStream() {
              this.port.postMessage({
                ended: true,
                currentTime,
                currentFrame,
                readOffset: this.readOffset,
                readIndex: this.readIndex,
                writeOffset: this.writeOffet,
                writeIndex: this.writeIndex,
                totalBytes: this.totalBytes,
              });
            }
            process(inputs, outputs) {
              const channels = outputs.flat();
              if (
                this.writeOffset >= this.totalBytes ||
                this.totalBytes === this.byteSize
              ) {
                console.log(this);
                this.endOfStream();
                return false;
              }

              const uint8 = new Uint8Array(512);
              try {
                for (let i = 0; i < 512; i++, this.writeOffset++) {
                  if (this.writeOffset === this.byteSize) {
                    break;
                  }
                  uint8[i] = this.memory[this.writeOffset];
                  // ++this.writeOffset;
                }
                const uint16 = new Uint16Array(uint8.buffer);
                // based on int16ToFloat32 function at https://stackoverflow.com/a/35248852
                for (let i = 0, j = 0, n = 1; i < uint16.length; i++) {
                  const int = uint16[i];
                  // If the high bit is on, then it is a negative number, and actually counts backwards.
                  const float =
                    int >= 0x8000 ? -(0x10000 - int) / 0x8000 : int / 0x7fff;
                  // interleave
                  channels[(n = ++n % 2)][!n ? j++ : j - 1] = float;
                }
              } catch (e) {
                console.error(e);
              }

              return true;
            }
          }
          // register processor in AudioWorkletGlobalScope
          function registerProcessor(name, processorCtor) {
            return `${processorCtor};\nregisterProcessor('${name}', ${processorCtor.name});`;
          }
          const worklet = URL.createObjectURL(
            new Blob(
              [
                registerProcessor(
                  'audio-worklet-native-file-stream',
                  AudioWorkletNativeFileStream
                ),
              ],
              { type: 'text/javascript' }
            )
          );
          const ac = new AudioContext({
            latencyHint: 1,
            sampleRate: 44100,
            numberOfChannels: 2,
          });
          ac.onstatechange = e => console.log(ac.state);
          if (ac.state === 'running') {
            await ac.suspend();
          }
          await ac.audioWorklet.addModule(worklet);
          const aw = new AudioWorkletNode(
            ac,
            'audio-worklet-native-file-stream',
            {
              numberOfInputs: 1,
              numberOfOutputs: 2,
              channelCount: 2,
              processorOptions: {
                totalBytes: 0,
                readIndex: 0,
                writeIndex: 0,
                readOffset: 0,
                writeOffset: 0,
                done: false,
                ended: false,
                started: false,
              },
            }
          );

          aw.onprocessorerror = e => {
            console.error(e);
            console.trace();
          };
          // comment MediaStream, MediaStreamTrack creation
          const msd = new MediaStreamAudioDestinationNode(ac);
          const { stream } = msd;
          const [track] = stream.getAudioTracks();
          aw.connect(msd);
          // aw.connect(ac.destination);
          const recorder = new MediaRecorder(stream);
          recorder.ondataavailable = e =>
            console.log(URL.createObjectURL(e.data));
          aw.port.onmessage = async e => {
            console.log(e.data);
            if (
              e.data.start &&
              ac.state === 'suspended'
            ) {
              await ac.resume();
              recorder.start();
              
            } else {
              if (recorder.state === 'recording') {
              
              recorder.stop();
              track.stop();
              aw.disconnect();
              msd.disconnect();
              await ac.suspend();
              console.log(track);
              // gc();
            }
           }
          };

          try {
            start.disabled = true;

            controller = new AbortController();
            signal = controller.signal;
            const { body: readable } = await fetch(
              'http://localhost:8000?start=true',
              {
                cache: 'no-store',
                signal,
              }
            );
            aw.port.postMessage(readable, [readable]);
          } catch (e) {
            console.warn(e);
          } finally {
             
          }
        };
        stop.onclick = e => {
          controller.abort();
          start.disabled = false;
        };
      </script>
    </body>
  </html>

[alvestrand]

Closing as "not identified as requiring any spec change".