import threading

import gi

import queue

from typing import List, Optional

import threading as _thread

Ensure Gst is imported first

gi.require_version('Gst', '1.0')

from gi.repository import Gst, GLib

from app.service.logs_service import logger

Initialize GStreamer once

Gst.init(None)

class WebmGstreamerConverter:

    """

    Thread-safe converter: WebM audio chunks -> (ogg/opus or raw pcm)

    - Runs GLib.MainLoop on a background thread

    - Uses appsink callbacks to push converted chunks into a thread-safe Queue

    - Uses an Event to signal finished/closed state (safer than peeking internal queue)

    """

    def init(self, pipeline: str, required_plugins: Optional[List[str]] = None):

        logger.info("Initializing WebmGstreamerConverter")

        # 0. Validate that required plugins exist (helpful in minimal containers)

        if required_plugins:

            for p in required_plugins:

                if not Gst.ElementFactory.find(p):

                    logger.error(f"GStreamer plugin '{p}' not available in the container")

                    raise RuntimeError(f"Missing GStreamer plugin: {p}")

        # 1. Pipeline Creation

        try:

            self.pipeline: Gst.Pipeline = Gst.parse_launch(pipeline)

            logger.debug(f"GStreamer pipeline created: {pipeline}")

        except Exception as e:

            logger.exception(f"Failed to create GStreamer pipeline: {e}")

            raise

        self.src = self.pipeline.get_by_name("src")

        self.sink = self.pipeline.get_by_name("sink")

        if not self.src or not self.sink:

            logger.error("Could not retrieve 'src' or 'sink' elements from pipeline. "

                         "Check pipeline string for 'name=src' and 'name=sink'.")

            raise RuntimeError("Invalid GStreamer pipeline elements")

        # 2. Message Bus Handler

        self.bus = self.pipeline.get_bus()

        self.bus.add_signal_watch()

        self.bus.connect("message", self._bus_call)

        logger.debug("GStreamer bus signal connected.")

        # 3. Appsink Connection

        logger.debug("Connecting appsink signal...")

        self.sink.connect("new-sample", self.on_new_sample)

        # 4. GLib MainLoop Thread

        self.loop = GLib.MainLoop()

        self.loop_thread = threading.Thread(target=self.loop.run, daemon=True)

        self.loop_thread.start()

        logger.info("GLib MainLoop started in background thread")

        # 5. Thread-safe Queue + finished event

        self._output_queue: queue.Queue = queue.Queue()

        self._finished_event = _thread.Event()

        # 6. Start Pipeline (safely wait for state transition)

        logger.info("Attempting to set pipeline state to PLAYING...")

        ret = self.pipeline.set_state(Gst.State.PLAYING)

        if ret == Gst.StateChangeReturn.FAILURE:

            logger.error("Failed to set pipeline state to PLAYING.")

            raise RuntimeError("Failed to start GStreamer pipeline.")

        elif ret == Gst.StateChangeReturn.ASYNC:

            # Wait up to 1 second for the state change to complete

            wait_ret, current, pending = self.pipeline.get_state(Gst.SECOND)

        else:

            wait_ret, current, pending = self.pipeline.get_state(0)

        logger.info(f"GStreamer pipeline state after transition attempt: current={current}, pending={pending}, ret={wait_ret}")

        if current != Gst.State.PLAYING:

            logger.error(f"Pipeline failed to reach PLAYING state. Current state: {current}")

            raise RuntimeError(f"GStreamer pipeline failed to start. Current state: {current}")

    # --- GStreamer message handler ---

    def _bus_call(self, bus: Gst.Bus, message: Gst.Message) -> bool:

        """Handle GStreamer EOS, ERROR, WARNING (runs on GLib thread)."""

        t = message.type

        if t == Gst.MessageType.EOS:

            logger.info("GStreamer End-Of-Stream reached.")

            # Put sentinel and set finished flag

            GLib.idle_add(self._output_queue.put, None)

            self._finished_event.set()

            try:

                # Quit mainloop gracefully

                self.loop.quit()

            except Exception:

                logger.debug("MainLoop quit raised exception (likely already quitting).")

        elif t == Gst.MessageType.ERROR:

            err, debug = message.parse_error()

            logger.error(f"[GStreamer ERROR] {err.message}")

            if debug:

                logger.debug(f"[GStreamer DEBUG INFO] {debug}")

            # Signal closure on error

            GLib.idle_add(self._output_queue.put, None)

            self._finished_event.set()

            try:

                self.loop.quit()

            except Exception:

                pass

        elif t == Gst.MessageType.WARNING:

            err, debug = message.parse_warning()

            logger.warning(f"[GStreamer WARNING] {err.message}")

            if debug:

                logger.debug(f"[GStreamer DEBUG INFO] {debug}")

        return True

    def on_new_sample(self, sink: Gst.Element) -> Gst.FlowReturn:

        """Callback when a new sample is available from appsink (runs on GLib thread)."""

        logger.debug("New sample received from appsink")

        try:

            sample: Gst.Sample = sink.emit("pull-sample")

            if not sample:

                return Gst.FlowReturn.OK

            buf: Gst.Buffer = sample.get_buffer()

            if not buf:

                logger.warning("Sample has no buffer")

                return Gst.FlowReturn.OK

            map_info = buf.map(Gst.MapFlags.READ)

            if map_info:

                # copy bytes to ensure the memory doesn't reference underlying mapped memory

                data = bytes(map_info.data)

                self._output_queue.put(data)

                buf.unmap(map_info)

                logger.debug(f"Pushed {len(data)} bytes into output queue")

            else:

                logger.warning("Failed to map buffer from sample")

        except Exception as e:

            logger.exception(f"Error in on_new_sample: {e}")

        return Gst.FlowReturn.OK

    def feed(self, webm_chunk: bytes):

        """Feed WebM data into the GStreamer pipeline (runs on asyncio thread)."""

        if not webm_chunk:

            logger.debug("feed() called with empty chunk")

            return

        try:

            buf = Gst.Buffer.new_allocate(None, len(webm_chunk), None)

            buf.fill(0, webm_chunk)

            ret = self.src.emit("push-buffer", buf)

            # Validate push result

            if ret != Gst.FlowReturn.OK:

                logger.warning(f"push-buffer returned non-OK flow return: {ret}")

            else:

                logger.debug(f"Pushed {len(webm_chunk)} bytes into pipeline")

        except Exception as e:

            logger.exception(f"Error feeding data into GStreamer pipeline: {e}")

    def read_ogg(self) -> bytes:

        """Retrieve accumulated converted data (called from asyncio thread)."""

        output_chunks: List[bytes] = []

        try:

            while True:

                # drain queue without blocking

                chunk = self._output_queue.get_nowait()

                if chunk is None:

                    # sentinel -> mark finished and stop reading

                    logger.debug("Sentinel received in read_ogg, marking finished and breaking loop.")

                    self._output_queue.task_done()

                    self._finished_event.set()

                    break

                output_chunks.append(chunk)

                self._output_queue.task_done()

        except queue.Empty:

            # queue empty: expected normal behavior

            pass

        if not output_chunks:

            logger.debug("read_ogg() called but no output available")

            return b""

        data = b"".join(output_chunks)

        logger.debug(f"Returning {len(data)} bytes of converted data")

        return data

    def is_finished(self) -> bool:

        """Checks if the conversion process has finished (by checking the event)."""

        return self._finished_event.is_set()

    def close(self, timeout: float = 2.0):

        """Gracefully stop the pipeline and ensure the background thread is joined."""

        logger.info("Closing WebmGstreamerConverter pipeline")

        try:

            # 1. Stop pushing more data

            try:

                self.src.emit("end-of-stream")

            except Exception:

                logger.debug("end-of-stream emit raised; pipeline may already be shutting down.")

            # 2. Signal finished (in case EOS didn't get delivered)

            self._finished_event.set()

            # 3. Stop the GLib MainLoop if running

            try:

                if self.loop.is_running():

                    self.loop.quit()

            except Exception:

                logger.debug("loop.quit() raised")

            # 4. Set pipeline to NULL state

            try:

                self.pipeline.set_state(Gst.State.NULL)

            except Exception:

                logger.exception("Error when setting pipeline to NULL")

            # 5. Join thread

            if self.loop_thread.is_alive():

                self.loop_thread.join(timeout=timeout)

            logger.info("Pipeline closed and GLib MainLoop stopped")

        except Exception as e:

            logger.exception(f"Error closing pipeline: {e}")

def get_webm_to_ogg_converter() -> WebmGstreamerConverter:

    logger.info("Creating WebM→Ogg converter (Opus re-encode)")

    required = ["matroskademux", "opusdec", "opusenc", "oggmux"]

    return WebmGstreamerConverter(

        "appsrc name=src is-live=true format=time do-timestamp=true ! "

        "matroskademux ! queue ! "

        "opusdec ! audioconvert ! audioresample ! "

        "opusenc ! "

        "oggmux name=muxer ! appsink name=sink emit-signals=true sync=false",

        required_plugins=required

    )

def get_webm_to_pcm_converter() -> WebmGstreamerConverter:

    logger.info("Creating WebM→PCM converter (S16LE, 16kHz)")

    required = ["matroskademux", "opusdec"]

    return WebmGstreamerConverter(

        "appsrc name=src is-live=true format=time do-timestamp=true ! "

        "matroskademux name=demux ! "

        "queue ! opusdec ! audioconvert ! audioresample ! "

        "audio/x-raw,format=S16LE,channels=1,rate=16000 ! "

        "appsink name=sink emit-signals=true sync=false",

        required_plugins=required

    )

Hello Nagalakshmi J,

Welcome to Microsoft Q&A.

Thank you for sharing the detailed explanation, code snippet, and environment context that’s very helpful.

I understand that you’re encountering segmentation faults during WebM → OGG/PCM conversions using GStreamer before streaming audio to Azure AI Speech for real-time transcription.

Segmentation faults usually point to a native-level crash within the GStreamer pipeline or one of its underlying plugins, often triggered by memory handling, thread synchronization, or pipeline teardown. Based on your setup, here are some recommendations and next steps that can help stabilize your application:

Potential Causes and Recommendations

Since you’ve tested with versions 1.20.3 and 1.24.2, please ensure that all required plugins are correctly installed and compatible with your container image. In some cases, switching versions (upgrading or downgrading) or reinstalling specific plugins can resolve native crashes.

Segmentation faults can often occur due to improper memory handling. Ensure that all buffers are properly mapped, copied, and unmapped (as you’ve already implemented). Avoid accessing or reusing unmapped memory references.

You’re already using a thread-safe queue between GLib and asyncio threads that’s great. Please also review whether any other shared objects (e.g., buffers or pipeline states) are accessed across threads without locking. GStreamer’s internal thread pool can sometimes lead to race conditions in such setups.

You can strengthen your _bus_call() method to capture detailed errors and warnings from GStreamer, logging stack traces and state changes before a crash. This can help trace the sequence leading up to the segmentation fault.

Try reproducing the issue with a minimal pipeline (e.g., only matroskademux → opusdec → appsink) and add elements step by step. This helps identify which stage introduces instability.

Run with higher verbosity to capture what happens before the crash:

GST_DEBUG=3,*GstPython*:5,*opus*:5,*matroska*:5

This can reveal which element or plugin fails.

Monitor CPU and memory usage within the AKS container to rule out resource exhaustion. Ensure you’re not hitting limits that could force abnormal termination.

Please refer this Use Codec Compressed Audio Input with GStreamer.

I Hope this helps. Do let me know if you have any further queries.

Thank you!

Hi Nagalakshmi J,

Did you get any chance to review the above response. Do let me know if you have any further queries.

Thank you!