How to fix av_interleaved_write_frame() broken pipe error in php

I have an issue using ffmpeg to stream audio and parse to google cloud speech to text in PHP.

It returns this output. I have tried delaying some part of the script, that did not solve it. I have also checked for similar questions. however, they are mostly in python and none of the solutions actually work for this.

  built with gcc 8 (GCC)
  cpudetect
  libavutil      56. 31.100 / 56. 31.100
  libavcodec     58. 54.100 / 58. 54.100
  libavformat    58. 29.100 / 58. 29.100
  libavdevice    58.  8.100 / 58.  8.100
  libavfilter     7. 57.100 /  7. 57.100
  libavresample   4.  0.  0 /  4.  0.  0
  libswscale      5.  5.100 /  5.  5.100
  libswresample   3.  5.100 /  3.  5.100
  libpostproc    55.  5.100 / 55.  5.100
Input #0, mp3, from 'https://npr-ice.streamguys1.com/live.mp3':
  Metadata:
    icy-br          : 96
    icy-description : NPR Program Stream
    icy-genre       : News and Talk
    icy-name        : NPR Program Stream
    icy-pub         : 0
    StreamTitle     :
  Duration: N/A, start: 0.000000, bitrate: 96 kb/s
    Stream #0:0: Audio: mp3, 32000 Hz, stereo, fltp, 96 kb/s
Stream mapping:
  Stream #0:0 -> #0:0 (mp3 (mp3float) -> pcm_s16le (native))
Press [q] to stop, [?] for help
Output #0, s16le, to 'pipe:':
  Metadata:
    icy-br          : 96
    icy-description : NPR Program Stream
    icy-genre       : News and Talk
    icy-name        : NPR Program Stream
    icy-pub         : 0
    StreamTitle     :
    encoder         : Lavf58.29.100
    Stream #0:0: Audio: pcm_s16le, 16000 Hz, mono, s16, 256 kb/s
    Metadata:
      encoder         : Lavc58.54.100 pcm_s16le
**av_interleaved_write_frame(): Broken pipe** 256.0kbits/s speed=1.02x
**Error writing trailer of pipe:: Broken pipe**
size=      54kB time=00:00:01.76 bitrate= 250.8kbits/s speed=0.465x
video:0kB audio:55kB subtitle:0kB other streams:0kB global headers:0kB muxing overhead: unknown
Conversion failed!

this is my PHP code

require_once 'vendor/autoload.php';
    
    $projectId = "xxx-45512";
    putenv('GOOGLE_APPLICATION_CREDENTIALS=' . __DIR__ . '/xxx-45512-be3eb805f1d7.json');
    
    // Database connection
    $pdo = new PDO('mysql:host=localhost;dbname=', '', '');
    $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
    
    $url = "https://npr-ice.streamguys1.com/live.mp3";
    
    $ffmpegCmd = "ffmpeg -re -i $url -acodec pcm_s16le -ac 1 -ar 16000 -f s16le -";
    
    $fp = popen($ffmpegCmd, "r");
    if (!$fp) {
        die("Failed to open FFmpeg stream.");
    }
    sleep(5);

    try {
        $client = new SpeechClient(['transport' => 'grpc', 'credentials' => json_decode(file_get_contents(getenv('GOOGLE_APPLICATION_CREDENTIALS')), true)]);
    } catch (Exception $e) {
        echo 'Error: ' . $e->getMessage(); 
        exit;
    }
    
    $recognitionConfig = new RecognitionConfig([
        'auto_decoding_config' => new AutoDetectDecodingConfig(),
        'language_codes' => ['en-US'],
        'model' => 'long',
    ]);
    
    $streamingConfig = new StreamingRecognitionConfig([
        'config' => $recognitionConfig,
    ]);
    
    $configRequest = new StreamingRecognizeRequest([
        'recognizer' => "projects/$projectId/locations/global/recognizers/_",
        'streaming_config' => $streamingConfig,
    ]);
    
    
    function streamAudio($fp)
    {
        while (!feof($fp)) {
            yield fread($fp, 4096);
        }
    }
    
    $responses = $client->streamingRecognize([
    'requests' => (function () use ($configRequest, $fp) {
            yield $configRequest; // Send initial config
            foreach (streamAudio($fp) as $audioChunk) {
                yield new StreamingRecognizeRequest(['audio' => $audioChunk]);
            }
        })()]
    );
    
    // $responses = $speechClient->streamingRecognize();
    // $responses->writeAll([$request,]);
    
    foreach ($responses as $response) {
        foreach ($response->getResults() as $result) {
            $transcript = $result->getAlternatives()[0]->getTranscript();
            // echo "Transcript: $transcript\n";
    
            // Insert into the database
            $stmt = $pdo->prepare("INSERT INTO transcriptions (transcript) VALUES (:transcript)");
            $stmt->execute(['transcript' => $transcript]);
        }
    }
    
    
    pclose($fp);
    $client->close();

I'm not sure what the issue is at this time.

UPDATE

I've done some more debugging and i have gotten the error to clear and to stream actually starts. However, I expect the audio to transcribe and update my database but instead I get this error when i close the stream

this is my updated code

    $handle = popen($ffmpegCommand, "r");

    try {
        $client = new SpeechClient(['transport' => 'grpc', 'credentials' => json_decode(file_get_contents(getenv('GOOGLE_APPLICATION_CREDENTIALS')), true)]);
    } catch (Exception $e) {
        echo 'Error: ' . $e->getMessage(); 
        exit;
    }
    
    try {
    $recognitionConfig = (new RecognitionConfig())
        ->setAutoDecodingConfig(new AutoDetectDecodingConfig())
        ->setLanguageCodes(['en-US'], ['en-UK'])
        ->setModel('long');
    } catch (Exception $e) {
        echo 'Error: ' . $e->getMessage(); 
        exit;
    }
    
    try {
        $streamConfig = (new StreamingRecognitionConfig())
        ->setConfig($recognitionConfig);
    } catch (Exception $e) {
        echo 'Error: ' . $e->getMessage();
        exit;
    }
    try {
        $configRequest = (new StreamingRecognizeRequest())
        ->setRecognizer("projects/$projectId/locations/global/recognizers/_")
        ->setStreamingConfig($streamConfig);
    } catch (Exception $e) {
        echo 'Error: ' . $e->getMessage(); 
        exit;
    }
    
    $stream = $client->streamingRecognize();
    $stream->write($configRequest);
    
    mysqli_query($conn, "INSERT INTO transcriptions (transcript) VALUES ('bef')");
    
    while (!feof($handle)) {
        $chunk = fread($handle, 25600);
        // printf('chunk: ' . $chunk);
        if ($chunk !== false) {
            try {
                $request = (new StreamingRecognizeRequest())
                        ->setAudio($chunk);
                    $stream->write($request);
            } catch (Exception $e) {
                printf('Errorc: ' . $e->getMessage());
            }
        }
    }
    
    
    $insr = json_encode($stream);
    mysqli_query($conn, "INSERT INTO transcriptions (transcript) VALUES ('$insr')");
    
    foreach ($stream->read() as $response) {
        mysqli_query($conn, "INSERT INTO transcriptions (transcript) VALUES ('loop1')");
        foreach ($response->getResults() as $result) {
            mysqli_query($conn, "INSERT INTO transcriptions (transcript) VALUES ('loop2')");
            foreach ($result->getAlternatives() as $alternative) {
                $trans = $alternative->getTranscript();
                mysqli_query($conn, "INSERT INTO transcriptions (transcript) VALUES ('$trans')");
            }
        }
    }
    
    pclose($handle);
    $stream->close();
    $client->close();```

Finally solved it.

To generate transcripts of an audio from a live source (e.g online radio, e.t.c) using Google cloud speech-to-text API V2 & PHP.

Follow the installation requirements on the google documentation page and then use this code below.

<?php

use Google\ApiCore\ApiException;
use Google\ApiCore\BidiStream;
use Google\Cloud\Speech\V2\AudioEncoding;
use Google\Cloud\Speech\V2\AutoDetectDecodingConfig;
use Google\Cloud\Speech\V2\Client\SpeechClient;
use Google\Cloud\Speech\V2\ExplicitDecodingConfig;
use Google\Cloud\Speech\V2\RecognitionConfig;
use Google\Cloud\Speech\V2\RecognitionFeatures;
use Google\Cloud\Speech\V2\StreamingRecognitionConfig;
use Google\Cloud\Speech\V2\StreamingRecognizeRequest;

$projectId = "xxx-4551";
putenv('GOOGLE_APPLICATION_CREDENTIALS=' . __DIR__ . '/xxx-4551-be3eb805f1d7.json');

$ffmpegInUrl = 'https://npr-ice.streamguys1.com/live.mp3';
$ffmpegCommand = implode(' ', [
    'ffmpeg', '-user_agent', 'FFmpeg', '-re', '-i', escapeshellarg($ffmpegInUrl),
    '-acodec', 'pcm_s16le', '-ac', '1', '-ar', '16000', '-f', 'wav', 'pipe:1'
]);
$handle = popen($ffmpegCommand, "r");

try {
    $client = new SpeechClient(['transport' => 'grpc', 'credentials' => json_decode(file_get_contents(getenv('GOOGLE_APPLICATION_CREDENTIALS')), true)]);
} catch (Exception $e) {
    echo 'Error: ' . $e->getMessage();
    exit;
}

try {
    $recognitionConfig = (new RecognitionConfig())
        ->setAutoDecodingConfig(new AutoDetectDecodingConfig())
        ->setLanguageCodes(['en-US'], ['en-UK'])
        ->setModel('long');
} catch (Exception $e) {
    echo 'Error: ' . $e->getMessage();
    exit;
}

try {
    $streamConfig = (new StreamingRecognitionConfig())
        ->setConfig($recognitionConfig);
} catch (Exception $e) {
    echo 'Error: ' . $e->getMessage();
    exit;
}
try {
    $configRequest = (new StreamingRecognizeRequest())
        ->setRecognizer("projects/$projectId/locations/global/recognizers/_")
        ->setStreamingConfig($streamConfig);
} catch (Exception $e) {
    echo 'Error: ' . $e->getMessage();
    exit;
}

$stream = $client->streamingRecognize();
$stream->write($configRequest);

while (!feof($handle)) {
    $chunk = fread($handle, 4096);
    if ($chunk !== false && strlen($chunk) > 0) {
        print_r('length_chunk_in');
        try {
            print_r('length_chunk_try: ' . strlen($chunk));
            $request = (new StreamingRecognizeRequest())
                ->setAudio($chunk);
            $stream->write($request);
        } catch (Exception $e) {
            var_dump('length_chunk_fail: ' . $e->getMessage());
        }
    } else {
        var_dump('length_chunk_null : ' . $handle);
    }
}


try {
    print_r('try_processing');
    foreach ($stream->closeWriteAndReadAll() as $response) {
        print_r('processing');
        foreach ($response->getResults() as $result) {
            foreach ($result->getAlternatives() as $alternative) {
                $trans = $alternative->getTranscript();
                mysqli_query($conn, "INSERT INTO transcriptions (transcript) VALUES ('$trans')");
            }
        }
    }
} catch (Google\ApiCore\ApiException $e) {
    if (strpos($e->getMessage(), "Stream timed out") !== false) {
        echo "Stream closed properly.\n";
    } else {
        throw $e;
    }
} finally {

}

pclose($handle);
$client->close();