Signal Handling

Use SignalEOS to gracefully shut down a pipeline on SIGINT (Ctrl+C) or SIGTERM instead of crashing.

Basic Usage

Inherit SignalEOS in your source element and check self.signaled_eos():

from dataclasses import dataclass
from sgn import SourceElement, SourcePad, Frame
from sgn.sources import SignalEOS


@dataclass
class StreamingSource(SourceElement, SignalEOS):
    """A source that runs until interrupted."""
    def new(self, pad: SourcePad) -> Frame:
        if self.signaled_eos():
            return Frame(EOS=True)
        data = produce_data()  # Your data generation logic
        return Frame(data=data)

Wrap the pipeline execution with the SignalEOS context manager:

from sgn import Pipeline
from sgn.sources import SignalEOS

p = Pipeline()
# ... build pipeline ...

with SignalEOS():
    p.run()

When a signal is received, signaled_eos() returns True. The source produces an EOS frame, which propagates through the pipeline, and all sinks mark EOS normally. The pipeline shuts down cleanly.

How It Works

1. On __enter__, SignalEOS replaces handlers for SIGINT and SIGTERM with a custom handler that records the signal.
2. On __exit__, the original signal handlers are restored.
3. signaled_eos() returns True if any handled signal has been received.

Built-in Support

NullSource already inherits SignalEOS, so it works out of the box:

from sgn import Pipeline, NullSource, NullSink
from sgn.sources import SignalEOS

src = NullSource(name="src", source_pad_names=["H1"], wait=1.0)
snk = NullSink(name="snk", sink_pad_names=["H1"])

p = Pipeline()
p.connect(src, snk)

with SignalEOS():
    p.run()
# Press Ctrl+C to stop

Combined with HTTPControl

HTTPControl inherits from SignalEOS, so using HTTPControl as a context manager also provides signal handling:

from sgn.control import HTTPControl

with HTTPControl():
    p.run()
# Both HTTP control and signal handling are active