I’ve tried to rewrit this early draft of hte scirpt multiple times and end up with the scirpt either freezing or crashing due to a broken pipe.

currently with this implementation it is freezing at this point in the log statements before the scirpt no longer runs.

full script and log statements attatched below.

Tue Aug 6 20:53:07 2024 – Setting up CUDA environment variable for GPU processing
Tue Aug 6 20:53:07 2024 – Setting GST_PLUGIN_PATH to /usr/lib/x86_64-linux-gnu/gstreamer-1.0
Tue Aug 6 20:53:07 2024 – Starting script
Tue Aug 6 20:53:07 2024 – Initializing GStreamer
Tue Aug 6 20:53:10 2024 – Enabled detailed GStreamer debug output
Tue Aug 6 20:53:10 2024 – Set GST_DEBUG_DUMP_DOT_DIR to /home/ubuntu/dot
Tue Aug 6 20:53:10 2024 – GStreamer pipeline element created successfully
Tue Aug 6 20:53:10 2024 – GStreamer source element created successfully
Tue Aug 6 20:53:10 2024 – GStreamer audiorate element created successfully
Tue Aug 6 20:53:10 2024 – GStreamer convert element created successfully
Tue Aug 6 20:53:10 2024 – GStreamer resample element created successfully
Tue Aug 6 20:53:10 2024 – GStreamer sink element created successfully
Tue Aug 6 20:53:10 2024 – GStreamer queue element created successfully
Tue Aug 6 20:53:10 2024 – Adding elements to GStreamer pipeline
Tue Aug 6 20:53:10 2024 – Successfully added source to the pipeline
Tue Aug 6 20:53:10 2024 – Successfully added queue to the pipeline
Tue Aug 6 20:53:10 2024 – Successfully added audiorate to the pipeline
Tue Aug 6 20:53:10 2024 – Successfully added convert to the pipeline
Tue Aug 6 20:53:10 2024 – Successfully added resample to the pipeline
Tue Aug 6 20:53:10 2024 – Successfully added sink to the pipeline
Tue Aug 6 20:53:10 2024 – Setting Twitch stream URL
Tue Aug 6 20:53:10 2024 – Adding signal watch to GStreamer bus
Tue Aug 6 20:53:10 2024 – Setting GStreamer pipeline to PLAYING state
Tue Aug 6 20:53:10 2024 – Pipeline state change is asynchronous
Tue Aug 6 20:53:12 2024 – New pad ‘src_0’ added to ‘source’
Tue Aug 6 20:53:12 2024 – Pad ‘src_0’ is not an audio pad, media type: ‘video/x-raw, width=(int)[ 48, 4096 ], height=(int)[ 16, 4096 ], framerate=(fraction)[ 0/1, 2147483647/1 ], format=(string){ NV12 }; video/x-raw(memory:GLMemory), width=(int)[ 48, 4096 ], height=(int)[ 16, 4096 ], framerate=(fraction)[ 0/1, 2147483647/1 ], format=(string){ NV12 }; video/x-raw(memory:CUDAMemory), width=(int)[ 48, 4096 ], height=(int)[ 16, 4096 ], framerate=(fraction)[ 0/1, 2147483647/1 ], format=(string){ NV12 }’
Tue Aug 6 20:53:12 2024 – New pad ‘src_1’ added to ‘source’
Tue Aug 6 20:53:12 2024 – Pad ‘src_1’ is an audio pad with media type ‘audio/x-raw, format=(string)S16LE, layout=(string)interleaved, rate=(int)[ 8000, 96000 ], channels=(int)[ 1, 8 ]’
Tue Aug 6 20:53:12 2024 – Successfully linked pad ‘src_1’ to audiorate sink pad

script freezes here

import subprocess
import numpy as np
import librosa
import logging
import os
import torch
from funasr import AutoModel
import signal
import time
import streamlink
import gi

# GStreamer setup
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib

# Constants
RATE = 16000
CHUNK = 1024
USERNAME = 'twitch_streamer_username'
MAX_RETRIES = 5
CHECKPOINT_DIR = './checkpoints'

# Twitch API credentials
CLIENT_ID = 'xxxxxxxxxxxxx'
CLIENT_SECRET = 'xxxxxxxxxxxxxx'

# Logging configuration
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def custom_sigpipe_handler(signum, frame):
    print(f"{time.ctime()} - SIGPIPE received: Broken pipe error", flush=True)
    logging.error("SIGPIPE received: Broken pipe error")

# Set up signal handling for SIGPIPE
signal.signal(signal.SIGPIPE, custom_sigpipe_handler)

# Ensure proper CUDA environment variable is set for GPU processing
os.environ["CUDA_VISIBLE_DEVICES"] = "0"
print(f"{time.ctime()} - Setting up CUDA environment variable for GPU processing", flush=True)

# GStreamer Plugin Pathway
plugin_path = "/usr/lib/x86_64-linux-gnu/gstreamer-1.0"
os.environ["GST_PLUGIN_PATH"] = plugin_path
print(f"{time.ctime()} - Setting GST_PLUGIN_PATH to {plugin_path}", flush=True)

# Custom SIGINT handler
def custom_sigint_handler(signum, frame):
    raise KeyboardInterrupt

signal.signal(signal.SIGINT, custom_sigint_handler)

# Function to get Twitch stream URL using Streamlink
def get_twitch_stream_url(username):
    try:
        session = streamlink.Streamlink()
        streams = session.streams(f"https://www.twitch.tv/{username}")
        if 'best' in streams:
            stream_url = streams['best'].url
            logging.info(f"Stream URL: {stream_url}")
            return stream_url
        else:
            logging.info("No playable streams found.")
            return None
    except Exception as e:
        logging.error(f"Error fetching Twitch stream URL: {e}")
        return None

# Error handling function for GStreamer
def on_error(bus, msg, pipeline):
    err, debug = msg.parse_error()
    print(f"{time.ctime()} - GStreamer error: {err}, {debug}", flush=True)
    logging.error(f"GStreamer error: {err}, {debug}")
    if "not-linked" in str(err):
        print(f"{time.ctime()} - Error: Element not linked properly", flush=True)
        logging.error("Element not linked properly")
    elif "capabilities" in str(err):
        print(f"{time.ctime()} - Error: Element capabilities issue", flush=True)
        logging.error("Element capabilities issue")
    pipeline.set_state(Gst.State.NULL)
    return

# Function to normalize audio
def normalize_audio(audio):
    print(f"{time.ctime()} - Normalizing audio", flush=True)
    try:
        return librosa.util.normalize(audio)
    except Exception as e:
        print(f"{time.ctime()} - Error normalizing audio: {e}", flush=True)
        return audio  # Return unnormalized audio in case of error

# Function to ensure finite values in audio buffer
def ensure_finite(audio):
    print(f"{time.ctime()} - Ensuring finite values in audio buffer", flush=True)
    try:
        return np.nan_to_num(audio)
    except Exception as e:
        print(f"{time.ctime()} - Error ensuring finite values: {e}", flush=True)
        return audio  # Return original audio in case of error

# Function to perform emotion recognition
def perform_emotion_recognition(audio_data, model):
    print(f"{time.ctime()} - Performing emotion recognition", flush=True)
    try:
        # Save the audio data to a temporary file
        temp_wav_file = "temp_audio.wav"
        librosa.output.write_wav(temp_wav_file, audio_data, RATE)
        # Perform emotion recognition using the funasr model
        res = model.generate(temp_wav_file, output_dir="./outputs", granularity="utterance", extract_embedding=False)
        return res
    except FileNotFoundError as e:
        print(f"{time.ctime()} - FileNotFoundError: {e}", flush=True)
    except librosa.util.exceptions.ParameterError as e:
        print(f"{time.ctime()} - librosa ParameterError: {e}", flush=True)
    except Exception as e:
        print(f"{time.ctime()} - Failed to perform emotion recognition: {e}", flush=True)
    return None

# Main function to stream and process audio
def stream_audio_from_twitch():
    print(f"{time.ctime()} - Initializing GStreamer", flush=True)
    Gst.init(None)

    # Enable detailed debug output
    os.environ["GST_DEBUG"] = "*:5"
    print(f"{time.ctime()} - Enabled detailed GStreamer debug output", flush=True)

    # Generate pipeline graphs
    os.environ["GST_DEBUG_DUMP_DOT_DIR"] = "/home/ubuntu/dot"
    print(f"{time.ctime()} - Set GST_DEBUG_DUMP_DOT_DIR to /home/ubuntu/dot", flush=True)

    # Create GStreamer elements
    elements = {
        "pipeline": Gst.Pipeline(),
        "source": Gst.ElementFactory.make("uridecodebin", "source"),
        "audiorate": Gst.ElementFactory.make("audiorate", "audiorate"),
        "convert": Gst.ElementFactory.make("audioconvert", "convert"),
        "resample": Gst.ElementFactory.make("audioresample", "resample"),
        "sink": Gst.ElementFactory.make("appsink", "sink"),
        "queue": Gst.ElementFactory.make("queue", "queue")
    }

    # Check each element and print specific error messages
    for name, element in elements.items():
        if not element:
            print(f"{time.ctime()} - Error: GStreamer {name} element could not be created", flush=True)
            logging.error(f"GStreamer {name} element could not be created.")
        else:
            print(f"{time.ctime()} - GStreamer {name} element created successfully", flush=True)
            logging.info(f"GStreamer {name} element created successfully")

    # If any element is missing, return early
    if not all(elements.values()):
        print(f"{time.ctime()} - Error: GStreamer elements could not be created", flush=True)
        logging.error("GStreamer elements could not be created.")
        return

    print(f"{time.ctime()} - Adding elements to GStreamer pipeline", flush=True)
    pipeline = elements["pipeline"]
    pipeline.set_state(Gst.State.NULL)  # Ensure pipeline is in NULL state

    for name in ["source", "queue", "audiorate", "convert", "resample", "sink"]:
        if pipeline.add(elements[name]):
            print(f"{time.ctime()} - Successfully added {name} to the pipeline", flush=True)
            logging.info(f"Successfully added {name} to the pipeline")
        else:
            print(f"{time.ctime()} - Failed to add {name} to the pipeline", flush=True)
            logging.error(f"Failed to add {name} to the pipeline")
            return  # Exit if any element fails to add

    # Correct handling of dynamic pads
    def on_pad_added(src, pad):
        print(f"{time.ctime()} - New pad '{pad.get_name()}' added to '{src.get_name()}'", flush=True)
        caps = pad.query_caps(None)
        media_type = caps.to_string()
        if "audio" in media_type:
            print(f"{time.ctime()} - Pad '{pad.get_name()}' is an audio pad with media type '{media_type}'", flush=True)
            sink_pad = elements["audiorate"].get_static_pad("sink")
            if pad.link(sink_pad) != Gst.PadLinkReturn.OK:
                print(f"{time.ctime()} - Failed to link pad '{pad.get_name()}' to audiorate sink pad", flush=True)
                logging.error(f"Failed to link pad '{pad.get_name()}' to audiorate sink pad")
            else:
                print(f"{time.ctime()} - Successfully linked pad '{pad.get_name()}' to audiorate sink pad", flush=True)
        else:
            print(f"{time.ctime()} - Pad '{pad.get_name()}' is not an audio pad, media type: '{media_type}'", flush=True)
            logging.info(f"Pad '{pad.get_name()}' is not an audio pad, media type: '{media_type}'")
            pad.set_active(False)  # Deactivate the pad to ignore video

    elements["source"].connect("pad-added", on_pad_added)

    print(f"{time.ctime()} - Setting Twitch stream URL", flush=True)
    stream_url = get_twitch_stream_url(USERNAME)
    if stream_url:
        elements["source"].set_property("uri", stream_url)
    else:
        print(f"{time.ctime()} - Failed to set Twitch stream URL", flush=True)
        return

    elements["sink"].set_property("emit-signals", True)

    print(f"{time.ctime()} - Adding signal watch to GStreamer bus", flush=True)
    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect("message::error", on_error, pipeline)

    def on_state_changed(bus, msg):
        if msg.type == Gst.MessageType.STATE_CHANGED:
            old_state, new_state, pending_state = msg.parse_state_changed()
            print(f"{time.ctime()} - Pipeline state changed from {old_state} to {new_state} (pending: {pending_state})", flush=True)
            logging.info(f"Pipeline state changed from {old_state} to {new_state} (pending: {pending_state})")
            if new_state == Gst.State.PLAYING:
                print(f"{time.ctime()} - Pipeline is now in PLAYING state", flush=True)

    bus.connect("message::state-changed", on_state_changed)

    def on_element_message(bus, msg):
        if msg.type == Gst.MessageType.ELEMENT:
            structure = msg.get_structure()
            if structure:
                logging.info(f"Element message: {structure.to_string()}")
                print(f"{time.ctime()} - Element message: {structure.to_string()}", flush=True)

    bus.connect("message::element", on_element_message)

    def on_buffering(bus, msg):
        if msg.type == Gst.MessageType.BUFFERING:
            percent = msg.parse_buffering()
            logging.info(f"Buffering {percent}%")
            print(f"{time.ctime()} - Buffering {percent}%", flush=True)

    bus.connect("message::buffering", on_buffering)

    print(f"{time.ctime()} - Setting GStreamer pipeline to PLAYING state", flush=True)
    ret = pipeline.set_state(Gst.State.PLAYING)
    if ret == Gst.StateChangeReturn.FAILURE:
        print(f"{time.ctime()} - Failed to set pipeline to PLAYING state", flush=True)
        # Capture detailed error messages
        bus.poll(Gst.MessageType.ERROR, Gst.CLOCK_TIME_NONE)
        return
    elif ret == Gst.StateChangeReturn.ASYNC:
        print(f"{time.ctime()} - Pipeline state change is asynchronous", flush=True)
        # Wait for the state change to complete
        state, current, pending = pipeline.get_state(Gst.CLOCK_TIME_NONE)
        print(f"{time.ctime()} - Pipeline state: {current}, Pending state: {pending}", flush=True)
        if current != Gst.State.PLAYING:
            print(f"{time.ctime()} - Pipeline failed to reach PLAYING state", flush=True)
            return
    print(f"{time.ctime()} - Pipeline is now in PLAYING state", flush=True)

    audio_data = bytearray()

    print(f"{time.ctime()} - Loading emotion recognition model", flush=True)
    try:
        model = AutoModel(model="iic/emotion2vec_plus_large")
    except torch.cuda.CudaError as e:
        print(f"{time.ctime()} - CUDA error while loading model: {e}", flush=True)
        return
    except Exception as e:
        print(f"{time.ctime()} - Failed to load emotion recognition model: {e}", flush=True)
        return

    last_sample_time = None

    def on_new_sample(sink):
        nonlocal audio_data, last_sample_time
        print(f"{time.ctime()} - New audio sample received", flush=True)
        try:
            start_time = time.time()
            sample = sink.emit("pull-sample")
            if not sample:
                print(f"{time.ctime()} - No sample received", flush=True)
                return Gst.FlowReturn.ERROR

            buf = sample.get_buffer()
            data = buf.extract_dup(0, buf.get_size())
            audio_data.extend(data)

            current_time = time.time()
            if last_sample_time is not None:
                interval = current_time - last_sample_time
                print(f"Time between samples: {interval:.6f} seconds", flush=True)
            last_sample_time = current_time

        except Exception as e:
            print(f"{time.ctime()} - Error extracting audio sample: {e}", flush=True)
            return Gst.FlowReturn.ERROR

        if len(audio_data) >= RATE * 5:
            print(f"{time.ctime()} - Processing 5 seconds of audio data", flush=True)
            try:
                audio_array = np.frombuffer(audio_data, dtype=np.float32)
                audio_array = ensure_finite(audio_array)
                audio_array = normalize_audio(audio_array)
                predictions = perform_emotion_recognition(audio_array, model)
                if predictions:
                    print(f"{time.ctime()} - Emotion predictions: {predictions}", flush=True)
                else:
                    print(f"{time.ctime()} - Failed to get emotion predictions", flush=True)
                audio_data = bytearray()
            except Exception as e:
                print(f"{time.ctime()} - Error processing audio data: {e}", flush=True)
                return Gst.FlowReturn.ERROR

        return Gst.FlowReturn.OK

    print(f"{time.ctime()} - Connecting new-sample signal to on_new_sample function", flush=True)
    elements["sink"].connect("new-sample", on_new_sample)

    print(f"{time.ctime()} - Starting GLib MainLoop", flush=True)
    loop = GLib.MainLoop()
    try:
        loop.run()
    except KeyboardInterrupt:
        print(f"{time.ctime()} - KeyboardInterrupt received, stopping GStreamer pipeline", flush=True)
    finally:
        print(f"{time.ctime()} - Setting GStreamer pipeline to NULL state", flush=True)
        pipeline.set_state(Gst.State.NULL)

if __name__ == "__main__":
    print(f"{time.ctime()} - Starting script", flush=True)
    try:
        stream_audio_from_twitch()
    except KeyboardInterrupt:
        print(f"{time.ctime()} - Script interrupted by user", flush=True)
    except Exception as e:
        print(f"{time.ctime()} - Unexpected error: {e}", flush=True)
    print(f"{time.ctime()} - Script finished", flush=True)type here

I have tried mostly focusing on hte gstreamer pipeline as i have never used gstreamer before but it could be possible that the problem lies elsewhere.