Skip to content

DataSource V2 API Reference

The datasource_v2 package provides a composable, dataclass-based API for creating gravitational wave data sources.

Main Entry Points

DataSource (Dispatcher)

sgnligo.sources.datasource_v2.DataSource

Dispatcher namespace for composed data sources.

Calling DataSource(data_source=..., ...) looks up the registered source class for data_source and returns an instance of it directly — the returned object IS a TSComposedSourceElement, not a wrapper, so pipeline.connect(source, sink) works without a .element indirection.

The returned instance also carries a data_source attribute set to the dispatch key for introspection.

Parameters:

Name Type Description Default
data_source

Source type identifier ("white", "gwdata-noise", ...).

 required
name

Name for the composed element.

 required
**kwargs

Forwarded to the selected source class.

 required

The classmethods on this class (from_argv, from_parser, create_cli_parser, list_sources, get_source_class) are a CLI-integration namespace — they also return composed-source instances directly.

Source code in sgnligo/sources/datasource_v2/datasource.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class DataSource:
    """Dispatcher namespace for composed data sources.

    Calling ``DataSource(data_source=..., ...)`` looks up the registered
    source class for ``data_source`` and returns an instance of it
    directly — the returned object IS a ``TSComposedSourceElement``,
    not a wrapper, so ``pipeline.connect(source, sink)`` works without
    a ``.element`` indirection.

    The returned instance also carries a ``data_source`` attribute set
    to the dispatch key for introspection.

    Args:
        data_source: Source type identifier ("white", "gwdata-noise", ...).
        name: Name for the composed element.
        **kwargs: Forwarded to the selected source class.

    The classmethods on this class (``from_argv``, ``from_parser``,
    ``create_cli_parser``, ``list_sources``, ``get_source_class``) are a
    CLI-integration namespace — they also return composed-source
    instances directly.
    """

    # Class metadata (for documentation only; not a registered source type)
    source_type: ClassVar[str] = "datasource"
    description: ClassVar[str] = "Unified data source dispatcher"

    def __new__(  # type: ignore[misc]
        cls,
        data_source: str,
        name: str = "datasource",
        **kwargs: Any,
    ) -> ComposedSourceBase:
        """Construct the appropriate composed source and return it directly."""
        inner_cls = get_composed_source_class(data_source)
        source = inner_cls(name=name, **kwargs)
        # Tag the dispatch key on the instance for introspection
        source.data_source = data_source  # type: ignore[attr-defined]
        return source

    # --- CLI Support ---

    @classmethod
    def create_cli_parser(
        cls,
        prog: Optional[str] = None,
        description: Optional[str] = None,
    ) -> argparse.ArgumentParser:
        """Create CLI argument parser with options for all registered sources.

        Use this when you need to add custom arguments to the parser.

        Example:
            >>> parser = DataSource.create_cli_parser()
            >>> parser.add_argument("--snr-threshold", type=float, default=8.0)
            >>> source, args = DataSource.from_parser(parser, name="pipeline")

        Returns:
            ArgumentParser configured with --data-source and all source options.
        """
        from sgnligo.sources.datasource_v2.cli import build_composed_cli_parser

        return build_composed_cli_parser(prog=prog, description=description)

    @classmethod
    def from_argv(
        cls,
        name: str = "datasource",
        argv: Optional[List[str]] = None,
    ) -> ComposedSourceBase:
        """Create a composed source from command line arguments.

        Parses sys.argv (or provided argv) and returns the appropriate
        composed source instance directly.

        Example:
            >>> # In a script called with:
            >>> # python script.py --data-source white --ifos H1
            >>> source = DataSource.from_argv(name="my_source")
            >>> pipeline.connect(source, sink)

        Args:
            name: Name for the composed element.
            argv: Command line arguments (defaults to sys.argv[1:]).

        Returns:
            A ``ComposedSourceBase`` subclass instance.
        """
        import sys

        from sgnligo.sources.datasource_v2.cli import (
            build_composed_cli_parser,
            check_composed_help_options,
            namespace_to_datasource_kwargs,
        )

        # Handle --list-sources and --help-source before parsing
        if check_composed_help_options(argv):
            sys.exit(0)

        parser = build_composed_cli_parser()
        args = parser.parse_args(argv)
        kwargs = namespace_to_datasource_kwargs(args, parser=parser)
        # cls(...) returns a ComposedSourceBase via __new__, but mypy doesn't
        # track that through the classmethod's `cls` — cast to help it.
        return cast(ComposedSourceBase, cls(name=name, **kwargs))

    @classmethod
    def from_parser(
        cls,
        parser: argparse.ArgumentParser,
        name: str = "datasource",
        argv: Optional[List[str]] = None,
    ) -> tuple[ComposedSourceBase, argparse.Namespace]:
        """Create a composed source from a custom argument parser.

        Use this when you've added custom arguments to the parser.
        Returns both the composed source and the parsed args so you can
        access your custom arguments.

        Example:
            >>> parser = DataSource.create_cli_parser()
            >>> parser.add_argument("--snr-threshold", type=float, default=8.0)
            >>> source, args = DataSource.from_parser(parser, name="pipeline")
            >>> print(f"SNR threshold: {args.snr_threshold}")

        Args:
            parser: ArgumentParser (from ``create_cli_parser`` plus any
                custom arguments).
            name: Name for the composed element.
            argv: Command line arguments (defaults to sys.argv[1:]).

        Returns:
            Tuple of (composed source instance, parsed args namespace).
        """
        import sys

        from sgnligo.sources.datasource_v2.cli import (
            check_composed_help_options,
            namespace_to_datasource_kwargs,
        )

        # Handle --list-sources and --help-source before parsing
        if check_composed_help_options(argv):
            sys.exit(0)

        args = parser.parse_args(argv)
        kwargs = namespace_to_datasource_kwargs(args, parser=parser)
        return cast(ComposedSourceBase, cls(name=name, **kwargs)), args

    @staticmethod
    def list_sources() -> List[str]:
        """List all available source types."""
        return list_composed_source_types()

    @staticmethod
    def get_source_class(source_type: str) -> Type[ComposedSourceBase]:
        """Get the source class for a given type."""
        return get_composed_source_class(source_type)

create_cli_parser(prog=None, description=None) classmethod

Create CLI argument parser with options for all registered sources.

Use this when you need to add custom arguments to the parser.

Example

parser = DataSource.create_cli_parser() parser.add_argument("--snr-threshold", type=float, default=8.0) source, args = DataSource.from_parser(parser, name="pipeline")

Returns:

Type Description
ArgumentParser

ArgumentParser configured with --data-source and all source options.
Source code in sgnligo/sources/datasource_v2/datasource.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@classmethod
def create_cli_parser(
    cls,
    prog: Optional[str] = None,
    description: Optional[str] = None,
) -> argparse.ArgumentParser:
    """Create CLI argument parser with options for all registered sources.

    Use this when you need to add custom arguments to the parser.

    Example:
        >>> parser = DataSource.create_cli_parser()
        >>> parser.add_argument("--snr-threshold", type=float, default=8.0)
        >>> source, args = DataSource.from_parser(parser, name="pipeline")

    Returns:
        ArgumentParser configured with --data-source and all source options.
    """
    from sgnligo.sources.datasource_v2.cli import build_composed_cli_parser

    return build_composed_cli_parser(prog=prog, description=description)

list_sources() staticmethod

List all available source types.

Source code in sgnligo/sources/datasource_v2/datasource.py 
196
197
198
199
@staticmethod
def list_sources() -> List[str]:
    """List all available source types."""
    return list_composed_source_types()

get_source_class(source_type) staticmethod

Get the source class for a given type.

Source code in sgnligo/sources/datasource_v2/datasource.py 
201
202
203
204
@staticmethod
def get_source_class(source_type: str) -> Type[ComposedSourceBase]:
    """Get the source class for a given type."""
    return get_composed_source_class(source_type)

CLI Support

sgnligo.sources.datasource_v2.cli

CLI support for composed data sources.

This module provides CLI argument parsing and help generation for the dataclass-based composed source classes.

CLI arguments are defined by mixin classes that sources inherit from. The build_composed_cli_parser() function aggregates CLI arguments from all registered sources by walking their MRO and collecting arguments from mixins.

Example

from sgnligo.sources.datasource_v2.cli import ( ... build_composed_cli_parser, ... check_composed_help_options, ... )

if check_composed_help_options(): ... sys.exit(0)

parser = build_composed_cli_parser() args = parser.parse_args()

build_composed_cli_parser(prog=None, description=None)

Build CLI parser by aggregating arguments from source mixins.

This function walks the MRO of all registered source classes and collects CLI arguments from mixins that implement the CLIMixinProtocol. Duplicate arguments (same arg defined by multiple mixins) raise an error.

Parameters:

Name Type Description Default
prog Optional[str]

Program name for help text

 None
description Optional[str]

Description for help text

 None

Returns:

Type Description
ArgumentParser

ArgumentParser configured with all source options

Raises:

Type Description
ValueError

If duplicate CLI arguments are detected
Source code in sgnligo/sources/datasource_v2/cli.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def build_composed_cli_parser(
    prog: Optional[str] = None,
    description: Optional[str] = None,
) -> argparse.ArgumentParser:
    """Build CLI parser by aggregating arguments from source mixins.

    This function walks the MRO of all registered source classes and
    collects CLI arguments from mixins that implement the CLIMixinProtocol.
    Duplicate arguments (same arg defined by multiple mixins) raise an error.

    Args:
        prog: Program name for help text
        description: Description for help text

    Returns:
        ArgumentParser configured with all source options

    Raises:
        ValueError: If duplicate CLI arguments are detected
    """
    parser = argparse.ArgumentParser(
        prog=prog,
        description=description or "Process gravitational wave data",
    )

    # Main dispatch option
    source_types = list_composed_source_types()
    parser.add_argument(
        "--data-source",
        required=True,
        choices=source_types,
        help="Type of data source to use",
    )

    # Help options (always available)
    parser.add_argument(
        "--list-sources",
        action="store_true",
        help="List available source types",
    )
    parser.add_argument(
        "--help-source",
        metavar="SOURCE",
        help="Show help for a specific source type",
    )

    # First pass: collect all CLI mixins from all registered sources
    # We need to process mixins with MORE args first (supersets before subsets)
    # to ensure variant mixins like StateVectorOptionsMixin (4 args) are processed
    # before StateVectorOnDictOnlyMixin (3 args), so all args get registered.
    cli_mixins: List[Type] = []
    seen_mixins: Set[Type] = set()

    for _source_type, cls in _COMPOSED_REGISTRY.items():
        for base in cls.__mro__:
            if base in seen_mixins:
                continue

            # Skip if not a CLI mixin (doesn't define add_cli_arguments directly)
            if "add_cli_arguments" not in base.__dict__:
                continue
            if "get_cli_arg_names" not in base.__dict__:
                continue  # pragma: no cover

            # Skip the protocol class itself
            if base is CLIMixinProtocol:
                continue  # pragma: no cover

            cli_mixins.append(base)
            seen_mixins.add(base)

    # Sort mixins by arg count descending - supersets first
    cli_mixins.sort(key=lambda m: len(m.get_cli_arg_names()), reverse=True)

    # Second pass: add arguments, skipping variant mixins with overlapping args.
    # Different sources use different variants of mixins (e.g., one source uses
    # StateVectorOptionsMixin with 4 args, another uses StateVectorOnDictOnlyMixin
    # with 3 args). By processing supersets first, we register all unique args.
    added_args: Set[str] = set()
    for mixin in cli_mixins:
        new_args = mixin.get_cli_arg_names()

        # Skip if this mixin shares any args with an already-processed mixin
        if new_args & added_args:
            continue

        added_args.update(new_args)
        mixin.add_cli_arguments(parser)

    return parser

check_composed_help_options(argv=None)

Check for --list-sources and --help-source before full parsing.

Call this before parse_args() to handle help options that don't require --data-source to be specified.

Parameters:

Name Type Description Default
argv Optional[List[str]]

Command line arguments (defaults to sys.argv[1:])

 None

Returns:

Type Description
bool

True if help was handled (caller should exit), False otherwise.
Source code in sgnligo/sources/datasource_v2/cli.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def check_composed_help_options(argv: Optional[List[str]] = None) -> bool:
    """Check for --list-sources and --help-source before full parsing.

    Call this before parse_args() to handle help options that don't require
    --data-source to be specified.

    Args:
        argv: Command line arguments (defaults to sys.argv[1:])

    Returns:
        True if help was handled (caller should exit), False otherwise.
    """
    if argv is None:
        argv = sys.argv[1:]

    if "--list-sources" in argv:
        print(format_composed_source_list())
        return True

    if "--help-source" in argv:
        try:
            idx = argv.index("--help-source")
            source_type = argv[idx + 1]
            if source_type in _COMPOSED_REGISTRY:
                print(format_composed_source_help(source_type))
                return True
            else:
                available = ", ".join(sorted(_COMPOSED_REGISTRY.keys()))
                print(f"Unknown source type '{source_type}'. Available: {available}")
                return True
        except IndexError:
            print("--help-source requires a source type argument")
            return True

    return False

namespace_to_datasource_kwargs(args, parser=None, ignore_unconsumed=None)

Convert argparse namespace to DataSource kwargs.

This function walks the MRO of the selected source class and calls process_cli_args() on each mixin to convert CLI arguments to field values.

Parameters:

Name Type Description Default
args Namespace

Parsed argparse namespace

 required
parser Optional[ArgumentParser]

The ArgumentParser that produced args. When provided, raises ValueError if the user set options that the selected source type does not use.

 None
ignore_unconsumed Optional[Set[str]]

Field names that the calling application consumes itself, exempting them from the strict unconsumed-arg check. Use when a tool legitimately repurposes a registered source option at the application level (e.g., sgnl-inspiral reads --injection-file for injection bookkeeping regardless of the selected data source).

 None

Returns:

Type Description
Dict[str, Any]

Dict of kwargs for DataSource
Source code in sgnligo/sources/datasource_v2/cli.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
def namespace_to_datasource_kwargs(
    args: argparse.Namespace,
    parser: Optional[argparse.ArgumentParser] = None,
    ignore_unconsumed: Optional[Set[str]] = None,
) -> Dict[str, Any]:
    """Convert argparse namespace to DataSource kwargs.

    This function walks the MRO of the selected source class and calls
    `process_cli_args()` on each mixin to convert CLI arguments to field values.

    Args:
        args: Parsed argparse namespace
        parser: The ArgumentParser that produced args. When provided,
            raises ValueError if the user set options that the selected
            source type does not use.
        ignore_unconsumed: Field names that the calling application consumes
            itself, exempting them from the strict unconsumed-arg check. Use
            when a tool legitimately repurposes a registered source option at
            the application level (e.g., sgnl-inspiral reads ``--injection-file``
            for injection bookkeeping regardless of the selected data source).

    Returns:
        Dict of kwargs for DataSource
    """
    kwargs: Dict[str, Any] = {
        "data_source": args.data_source,
    }

    # Get the source class to walk its MRO for process_cli_args
    source_type = args.data_source
    if source_type in _COMPOSED_REGISTRY:
        cls = _COMPOSED_REGISTRY[source_type]
        processed_classes: Set[Type] = set()

        # Walk MRO and call process_cli_args on each mixin
        for base in cls.__mro__:
            if base in processed_classes:
                continue  # pragma: no cover

            if not hasattr(base, "process_cli_args"):
                continue

            if base is CLIMixinProtocol:
                continue  # pragma: no cover

            mixin_kwargs = base.process_cli_args(args)
            kwargs.update(mixin_kwargs)

            processed_classes.add(base)

        # Raise if the user provided args that this source type ignores
        if parser is not None:
            check_unconsumed_cli_args(
                args,
                selected_type=source_type,
                selected_cls=cls,
                registry=_COMPOSED_REGISTRY,
                protocol_cls=CLIMixinProtocol,
                type_flag="--data-source",
                parser=parser,
                ignore=ignore_unconsumed,
            )

    return kwargs

format_composed_source_help(source_type)

Generate detailed help for a specific source type.

Parameters:

Name Type Description Default
source_type str

The source type to show help for

 required

Returns:

Type Description
str

Formatted help string
Source code in sgnligo/sources/datasource_v2/cli.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def format_composed_source_help(source_type: str) -> str:
    """Generate detailed help for a specific source type.

    Args:
        source_type: The source type to show help for

    Returns:
        Formatted help string
    """
    cls = get_composed_source_class(source_type)

    lines = [
        f"usage: prog --data-source {source_type} [options]",
        "",
        f"{source_type}: {cls.description}",
        "",
    ]

    # Required fields
    required = get_source_required_fields(cls)
    if required:
        lines.append("Required Options:")
        for name in required:
            cli_name = name.replace("_", "-")
            lines.append(f"  --{cli_name}")
        lines.append("")

    # Optional fields
    optional = get_source_optional_fields(cls)
    if optional:
        lines.append("Optional Options:")
        for name, default in optional.items():
            cli_name = name.replace("_", "-")
            if default is False:
                lines.append(f"  --{cli_name}")
            elif default is not None:
                lines.append(f"  --{cli_name} (default: {default})")
            else:
                lines.append(f"  --{cli_name}")
        lines.append("")

    # Add docstring notes if available
    if cls.__doc__:
        # Extract just the description part (first paragraph)
        doc_lines = cls.__doc__.strip().split("\n\n")[0].split("\n")
        if doc_lines:
            lines.append("Description:")
            for doc_line in doc_lines:
                lines.append(f"  {doc_line.strip()}")

    return "\n".join(lines)

format_composed_source_list()

Generate list of all available sources.

Returns:

Type Description
str

Formatted string listing all sources
Source code in sgnligo/sources/datasource_v2/cli.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def format_composed_source_list() -> str:
    """Generate list of all available sources.

    Returns:
        Formatted string listing all sources
    """
    lines = ["Available data sources:", ""]

    for source_type in sorted(_COMPOSED_REGISTRY.keys()):
        cls = _COMPOSED_REGISTRY[source_type]
        lines.append(f"  {source_type:30} {cls.description}")

    lines.append("")
    lines.append("Use --help-source <name> for detailed options.")
    lines.append("Use --real-time flag with supported sources for real-time mode.")
    return "\n".join(lines)

Registry

sgnligo.sources.datasource_v2.composed_registry

Registry for dataclass-based composed source classes.

This module provides registration and lookup for the new dataclass-based source classes that inherit from ComposedSourceBase.

Example

from sgnligo.sources.datasource_v2.composed_registry import ( ... register_composed_source, ... get_composed_source_class, ... )

@register_composed_source @dataclass class MySource(ComposedSourceBase): ... source_type: ClassVar[str] = "my-source" ... ...

register_composed_source(cls)

Decorator to register a composed source class.

Parameters:

Name Type Description Default
cls Type[ComposedSourceBase]

The composed source class to register

 required

Returns:

Type Description
Type[ComposedSourceBase]

The same class (unchanged)

Raises:

Type Description
ValueError

If the source_type is empty or already registered
Example

@register_composed_source @dataclass class WhiteSource(ComposedSourceBase): source_type: ClassVar[str] = "white" ...

Source code in sgnligo/sources/datasource_v2/composed_registry.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def register_composed_source(
    cls: Type[ComposedSourceBase],
) -> Type[ComposedSourceBase]:
    """Decorator to register a composed source class.

    Args:
        cls: The composed source class to register

    Returns:
        The same class (unchanged)

    Raises:
        ValueError: If the source_type is empty or already registered

    Example:
        @register_composed_source
        @dataclass
        class WhiteSource(ComposedSourceBase):
            source_type: ClassVar[str] = "white"
            ...
    """
    source_type = cls.source_type
    if not source_type:
        raise ValueError(f"Class {cls.__name__} must define source_type")
    if source_type in _COMPOSED_REGISTRY:
        raise ValueError(
            f"Source type '{source_type}' is already registered "
            f"(by {_COMPOSED_REGISTRY[source_type].__name__})"
        )
    _COMPOSED_REGISTRY[source_type] = cls
    return cls

get_composed_source_class(source_type)

Get the composed source class for a given type string.

Parameters:

Name Type Description Default
source_type str

The source type identifier

 required

Returns:

Type Description
Type[ComposedSourceBase]

The composed source class

Raises:

Type Description
ValueError

If the source type is not registered
Source code in sgnligo/sources/datasource_v2/composed_registry.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def get_composed_source_class(source_type: str) -> Type[ComposedSourceBase]:
    """Get the composed source class for a given type string.

    Args:
        source_type: The source type identifier

    Returns:
        The composed source class

    Raises:
        ValueError: If the source type is not registered
    """
    if source_type not in _COMPOSED_REGISTRY:
        available = ", ".join(sorted(_COMPOSED_REGISTRY.keys()))
        raise ValueError(f"Unknown source type '{source_type}'. Available: {available}")
    return _COMPOSED_REGISTRY[source_type]

list_composed_source_types()

List all registered composed source types.

Returns:

Type Description
List[str]

Sorted list of registered source type names
Source code in sgnligo/sources/datasource_v2/composed_registry.py 
81
82
83
84
85
86
87
def list_composed_source_types() -> List[str]:
    """List all registered composed source types.

    Returns:
        Sorted list of registered source type names
    """
    return sorted(_COMPOSED_REGISTRY.keys())

get_composed_registry()

Get the full composed source registry.

Returns:

Type Description
Dict[str, Type[ComposedSourceBase]]

Dict mapping source type to class
Source code in sgnligo/sources/datasource_v2/composed_registry.py 
90
91
92
93
94
95
96
def get_composed_registry() -> Dict[str, Type[ComposedSourceBase]]:
    """Get the full composed source registry.

    Returns:
        Dict mapping source type to class
    """
    return _COMPOSED_REGISTRY.copy()

Source Classes

Fake Sources

sgnligo.sources.datasource_v2.sources.fake

Fake signal source classes (white, sin, impulse).

These sources generate synthetic test signals for pipeline development and testing without requiring real detector data.

Supports both offline (batch) and real-time modes via the real_time flag: - Offline (default): Requires start and end (or duration) - Real-time: GPS times optional, generates data synchronized with wall clock

Example

Offline mode (default)

source = WhiteComposedSource( ... name="noise", ... ifos=["H1", "L1"], ... channel_dict={"H1": "FAKE-STRAIN", "L1": "FAKE-STRAIN"}, ... sample_rate=4096, ... start=1000, ... end=1010, ... )

Real-time mode

source = WhiteComposedSource( ... name="noise", ... ifos=["H1"], ... channel_dict={"H1": "FAKE-STRAIN"}, ... sample_rate=4096, ... real_time=True, ... )

WhiteComposedSource dataclass

Bases: FakeSourceBase


              flowchart TD
              sgnligo.sources.datasource_v2.sources.fake.WhiteComposedSource[WhiteComposedSource]
              sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase[FakeSourceBase]
              sgnligo.sources.composed_base.ComposedSourceBase[ComposedSourceBase]
              sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin[ChannelOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin[SampleRateOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin[GPSOptionsFlexibleMixin]
              sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin[SegmentsOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin[VerboseOptionsMixin]

                              sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase --> sgnligo.sources.datasource_v2.sources.fake.WhiteComposedSource
                                sgnligo.sources.composed_base.ComposedSourceBase --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                



              click sgnligo.sources.datasource_v2.sources.fake.WhiteComposedSource href "" "sgnligo.sources.datasource_v2.sources.fake.WhiteComposedSource"
              click sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase href "" "sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase"
              click sgnligo.sources.composed_base.ComposedSourceBase href "" "sgnligo.sources.composed_base.ComposedSourceBase"
              click sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin"

Gaussian white noise source.

Generates uncorrelated Gaussian white noise for each IFO channel. Useful for basic pipeline testing where spectral characteristics don't matter.

Supports both offline and real-time modes: - Offline (default): Specify start and end (or duration) - Real-time: Set real_time=True, GPS times become optional

Example

Offline mode

source = WhiteComposedSource( ... name="noise", ... ifos=["H1", "L1"], ... channel_dict={"H1": "FAKE-STRAIN", "L1": "FAKE-STRAIN"}, ... sample_rate=4096, ... start=1000, ... end=1010, ... )

Real-time mode

source = WhiteComposedSource( ... name="noise", ... ifos=["H1"], ... channel_dict={"H1": "FAKE-STRAIN"}, ... sample_rate=4096, ... real_time=True, ... )

Source code in sgnligo/sources/datasource_v2/sources/fake.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
@register_composed_source
@dataclass(kw_only=True)
class WhiteComposedSource(FakeSourceBase):
    """Gaussian white noise source.

    Generates uncorrelated Gaussian white noise for each IFO channel.
    Useful for basic pipeline testing where spectral characteristics
    don't matter.

    Supports both offline and real-time modes:
    - Offline (default): Specify start and end (or duration)
    - Real-time: Set real_time=True, GPS times become optional

    Example:
        >>> # Offline mode
        >>> source = WhiteComposedSource(
        ...     name="noise",
        ...     ifos=["H1", "L1"],
        ...     channel_dict={"H1": "FAKE-STRAIN", "L1": "FAKE-STRAIN"},
        ...     sample_rate=4096,
        ...     start=1000,
        ...     end=1010,
        ... )
        >>>
        >>> # Real-time mode
        >>> source = WhiteComposedSource(
        ...     name="noise",
        ...     ifos=["H1"],
        ...     channel_dict={"H1": "FAKE-STRAIN"},
        ...     sample_rate=4096,
        ...     real_time=True,
        ... )
    """

    source_type: ClassVar[str] = "white"
    description: ClassVar[str] = "Gaussian white noise"
    signal_type: ClassVar[str] = "white"

SinComposedSource dataclass

Bases: FakeSourceBase


              flowchart TD
              sgnligo.sources.datasource_v2.sources.fake.SinComposedSource[SinComposedSource]
              sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase[FakeSourceBase]
              sgnligo.sources.composed_base.ComposedSourceBase[ComposedSourceBase]
              sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin[ChannelOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin[SampleRateOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin[GPSOptionsFlexibleMixin]
              sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin[SegmentsOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin[VerboseOptionsMixin]

                              sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase --> sgnligo.sources.datasource_v2.sources.fake.SinComposedSource
                                sgnligo.sources.composed_base.ComposedSourceBase --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                



              click sgnligo.sources.datasource_v2.sources.fake.SinComposedSource href "" "sgnligo.sources.datasource_v2.sources.fake.SinComposedSource"
              click sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase href "" "sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase"
              click sgnligo.sources.composed_base.ComposedSourceBase href "" "sgnligo.sources.composed_base.ComposedSourceBase"
              click sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin"

Sinusoidal test signal source.

Generates a sinusoidal signal for each IFO channel. Useful for testing frequency-domain processing.

Supports both offline and real-time modes.

Example

source = SinComposedSource( ... name="sine", ... ifos=["H1"], ... channel_dict={"H1": "FAKE-STRAIN"}, ... sample_rate=4096, ... start=1000, ... end=1010, ... )

Source code in sgnligo/sources/datasource_v2/sources/fake.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
@register_composed_source
@dataclass(kw_only=True)
class SinComposedSource(FakeSourceBase):
    """Sinusoidal test signal source.

    Generates a sinusoidal signal for each IFO channel.
    Useful for testing frequency-domain processing.

    Supports both offline and real-time modes.

    Example:
        >>> source = SinComposedSource(
        ...     name="sine",
        ...     ifos=["H1"],
        ...     channel_dict={"H1": "FAKE-STRAIN"},
        ...     sample_rate=4096,
        ...     start=1000,
        ...     end=1010,
        ... )
    """

    source_type: ClassVar[str] = "sin"
    description: ClassVar[str] = "Sinusoidal test signal"
    signal_type: ClassVar[str] = "sin"

ImpulseComposedSource dataclass

Bases: FakeSourceBase, ImpulsePositionOptionsMixin


              flowchart TD
              sgnligo.sources.datasource_v2.sources.fake.ImpulseComposedSource[ImpulseComposedSource]
              sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase[FakeSourceBase]
              sgnligo.sources.composed_base.ComposedSourceBase[ComposedSourceBase]
              sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin[ChannelOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin[SampleRateOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin[GPSOptionsFlexibleMixin]
              sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin[SegmentsOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin[VerboseOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.ImpulsePositionOptionsMixin[ImpulsePositionOptionsMixin]

                              sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase --> sgnligo.sources.datasource_v2.sources.fake.ImpulseComposedSource
                                sgnligo.sources.composed_base.ComposedSourceBase --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                
                sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase
                

                sgnligo.sources.datasource_v2.cli_mixins.ImpulsePositionOptionsMixin --> sgnligo.sources.datasource_v2.sources.fake.ImpulseComposedSource
                


              click sgnligo.sources.datasource_v2.sources.fake.ImpulseComposedSource href "" "sgnligo.sources.datasource_v2.sources.fake.ImpulseComposedSource"
              click sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase href "" "sgnligo.sources.datasource_v2.sources.fake.FakeSourceBase"
              click sgnligo.sources.composed_base.ComposedSourceBase href "" "sgnligo.sources.composed_base.ComposedSourceBase"
              click sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.SampleRateOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.ImpulsePositionOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.ImpulsePositionOptionsMixin"

Impulse test signal source.

Generates an impulse signal (single spike) for each IFO channel. Useful for testing impulse response.

Supports both offline and real-time modes.

Fields inherited from mixins

impulse_position: Sample index for impulse (-1 for random)

Example

source = ImpulseComposedSource( ... name="impulse", ... ifos=["H1"], ... channel_dict={"H1": "FAKE-STRAIN"}, ... sample_rate=4096, ... start=1000, ... end=1010, ... impulse_position=100, ... )

Source code in sgnligo/sources/datasource_v2/sources/fake.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
@register_composed_source
@dataclass(kw_only=True)
class ImpulseComposedSource(FakeSourceBase, ImpulsePositionOptionsMixin):
    """Impulse test signal source.

    Generates an impulse signal (single spike) for each IFO channel.
    Useful for testing impulse response.

    Supports both offline and real-time modes.

    Fields inherited from mixins:
        impulse_position: Sample index for impulse (-1 for random)

    Example:
        >>> source = ImpulseComposedSource(
        ...     name="impulse",
        ...     ifos=["H1"],
        ...     channel_dict={"H1": "FAKE-STRAIN"},
        ...     sample_rate=4096,
        ...     start=1000,
        ...     end=1010,
        ...     impulse_position=100,
        ... )
    """

    source_type: ClassVar[str] = "impulse"
    description: ClassVar[str] = "Impulse test signal"
    signal_type: ClassVar[str] = "impulse"

GWData Noise Sources

sgnligo.sources.datasource_v2.sources.gwdata_noise

GWData noise composed source class.

Generates colored Gaussian noise with realistic LIGO PSDs, suitable for testing and development without real detector data.

Supports both offline (batch) and real-time modes via the real_time flag: - Offline (default): Requires start and end (or duration) - Real-time: GPS times optional, generates data synchronized with wall clock

Example

Offline mode

source = GWDataNoiseComposedSource( ... name="noise", ... ifos=["H1", "L1"], ... channel_dict={"H1": "FAKE-STRAIN", "L1": "FAKE-STRAIN"}, ... start=1000, ... end=1010, ... )

Real-time mode

source = GWDataNoiseComposedSource( ... name="noise", ... ifos=["H1"], ... channel_dict={"H1": "FAKE-STRAIN"}, ... real_time=True, ... )

GWDataNoiseComposedSource dataclass

Bases: ComposedSourceBase, ChannelOptionsMixin, GPSOptionsFlexibleMixin, StateVectorOnDictOnlyMixin, VerboseOptionsMixin


              flowchart TD
              sgnligo.sources.datasource_v2.sources.gwdata_noise.GWDataNoiseComposedSource[GWDataNoiseComposedSource]
              sgnligo.sources.composed_base.ComposedSourceBase[ComposedSourceBase]
              sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin[ChannelOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin[GPSOptionsFlexibleMixin]
              sgnligo.sources.datasource_v2.cli_mixins.StateVectorOnDictOnlyMixin[StateVectorOnDictOnlyMixin]
              sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin[VerboseOptionsMixin]

                              sgnligo.sources.composed_base.ComposedSourceBase --> sgnligo.sources.datasource_v2.sources.gwdata_noise.GWDataNoiseComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin --> sgnligo.sources.datasource_v2.sources.gwdata_noise.GWDataNoiseComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin --> sgnligo.sources.datasource_v2.sources.gwdata_noise.GWDataNoiseComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.StateVectorOnDictOnlyMixin --> sgnligo.sources.datasource_v2.sources.gwdata_noise.GWDataNoiseComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin --> sgnligo.sources.datasource_v2.sources.gwdata_noise.GWDataNoiseComposedSource
                


              click sgnligo.sources.datasource_v2.sources.gwdata_noise.GWDataNoiseComposedSource href "" "sgnligo.sources.datasource_v2.sources.gwdata_noise.GWDataNoiseComposedSource"
              click sgnligo.sources.composed_base.ComposedSourceBase href "" "sgnligo.sources.composed_base.ComposedSourceBase"
              click sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.StateVectorOnDictOnlyMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.StateVectorOnDictOnlyMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin"

Colored Gaussian noise source with optional state vector gating.

Generates colored Gaussian noise with LIGO PSD. Supports both offline and real-time modes, with optional segment-based state vector gating.

Fields inherited from mixins

ifos: List of detector prefixes (from ChannelOptionsMixin) channel_dict: Dict mapping IFO to channel name (from ChannelOptionsMixin) start: GPS start time (from GPSOptionsFlexibleMixin) end: GPS end time (from GPSOptionsFlexibleMixin) duration: Duration in seconds, alternative to end (from GPSOptionsFlexibleMixin) real_time: Enable real-time mode (from GPSOptionsFlexibleMixin) state_vector_on_dict: Bitmask dict (from StateVectorOnDictOnlyMixin) state_segments_file: State segments file (from StateVectorOnDictOnlyMixin) state_sample_rate: State vector sample rate (from StateVectorOnDictOnlyMixin) verbose: Enable verbose output (from VerboseOptionsMixin)

Example

Offline mode

source = GWDataNoiseComposedSource( ... name="noise", ... ifos=["H1", "L1"], ... channel_dict={"H1": "FAKE-STRAIN", "L1": "FAKE-STRAIN"}, ... start=1000, ... end=1010, ... )

Real-time mode

source = GWDataNoiseComposedSource( ... name="noise", ... ifos=["H1"], ... channel_dict={"H1": "FAKE-STRAIN"}, ... real_time=True, ... )

Source code in sgnligo/sources/datasource_v2/sources/gwdata_noise.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
@register_composed_source
@dataclass(kw_only=True)
class GWDataNoiseComposedSource(
    ComposedSourceBase,
    ChannelOptionsMixin,
    GPSOptionsFlexibleMixin,
    StateVectorOnDictOnlyMixin,
    VerboseOptionsMixin,
):
    """Colored Gaussian noise source with optional state vector gating.

    Generates colored Gaussian noise with LIGO PSD. Supports both offline
    and real-time modes, with optional segment-based state vector gating.

    Fields inherited from mixins:
        ifos: List of detector prefixes (from ChannelOptionsMixin)
        channel_dict: Dict mapping IFO to channel name (from ChannelOptionsMixin)
        start: GPS start time (from GPSOptionsFlexibleMixin)
        end: GPS end time (from GPSOptionsFlexibleMixin)
        duration: Duration in seconds, alternative to end (from GPSOptionsFlexibleMixin)
        real_time: Enable real-time mode (from GPSOptionsFlexibleMixin)
        state_vector_on_dict: Bitmask dict (from StateVectorOnDictOnlyMixin)
        state_segments_file: State segments file (from StateVectorOnDictOnlyMixin)
        state_sample_rate: State vector sample rate (from StateVectorOnDictOnlyMixin)
        verbose: Enable verbose output (from VerboseOptionsMixin)

    Example:
        >>> # Offline mode
        >>> source = GWDataNoiseComposedSource(
        ...     name="noise",
        ...     ifos=["H1", "L1"],
        ...     channel_dict={"H1": "FAKE-STRAIN", "L1": "FAKE-STRAIN"},
        ...     start=1000,
        ...     end=1010,
        ... )
        >>>
        >>> # Real-time mode
        >>> source = GWDataNoiseComposedSource(
        ...     name="noise",
        ...     ifos=["H1"],
        ...     channel_dict={"H1": "FAKE-STRAIN"},
        ...     real_time=True,
        ... )
    """

    # Class metadata
    source_type: ClassVar[str] = "gwdata-noise"
    description: ClassVar[str] = "Colored Gaussian noise with LIGO PSD"

    def _validate(self) -> None:
        """Validate parameters."""
        # Validate GPS options based on real_time mode
        self._validate_gps_options()

        # Validate channel_dict keys match ifos
        if set(self.channel_dict.keys()) != set(self.ifos):
            raise ValueError("channel_dict keys must match ifos")

        # Validate state segments file
        if self.state_segments_file is not None:
            if not os.path.exists(self.state_segments_file):
                raise ValueError(
                    f"State segments file does not exist: {self.state_segments_file}"
                )

        # Validate state_vector_on_dict
        if self.state_vector_on_dict is not None:
            if set(self.state_vector_on_dict.keys()) != set(self.ifos):
                raise ValueError("state_vector_on_dict keys must match ifos")

    def _load_state_segments(
        self,
        end_time: Optional[float],
    ) -> Tuple[Optional[Tuple[Tuple[int, int], ...]], Optional[Tuple[int, ...]]]:
        """Load state segments from file or create defaults.

        Returns:
            Tuple of (segments, values) where segments is a tuple of (start_ns, end_ns)
            pairs and values is a tuple of state vector values. Returns (None, None)
            if state vector gating is not configured.
        """
        if self.state_vector_on_dict is None:
            return None, None

        if self.state_segments_file is not None:
            state_segments, state_values = read_segments_and_values_from_file(
                self.state_segments_file, self.verbose
            )
        else:
            # Default: single segment covering entire time range with value 3
            if self.start is not None:
                start_ns = int(self.start * 1e9)
                if end_time is not None:
                    end_ns = int(end_time * 1e9)
                else:
                    # For real-time mode without end time
                    end_ns = int(np.iinfo(np.int32).max * 1e9)
                state_segments = ((start_ns, end_ns),)
                state_values = (3,)  # Default: bits 0 and 1 set
                if self.verbose:
                    print("Using default state segments: single segment with value 3")
            else:
                raise ValueError(
                    "Must provide either state_segments_file or start "
                    "when using state vector gating"
                )

        return state_segments, state_values

    def _build(self) -> None:
        """Build the GWData noise source."""
        # Get computed end time (may be calculated from duration)
        end_time = self._get_computed_end()

        # Build full channel names for internal use
        full_channel_dict = {
            ifo: f"{ifo}:{self.channel_dict[ifo]}" for ifo in self.ifos
        }

        # Create the noise source
        noise_source = GWDataNoiseSource(
            name=f"{self.name}_noise",
            channel_dict=full_channel_dict,
            start=self.start,
            end=end_time,
            real_time=self.real_time,
            verbose=self.verbose,
        )

        # Check if we need state vector gating
        if self.state_vector_on_dict is not None:
            state_segments, state_values = self._load_state_segments(end_time)
            assert state_segments is not None
            assert state_values is not None

            # Determine end time for SegmentSource (doesn't support None)
            seg_end = (
                end_time if end_time is not None else float(np.iinfo(np.int32).max)
            )

            for ifo in self.ifos:
                full_channel = full_channel_dict[ifo]

                # Create segment source for state vector
                state_source = SegmentSource(
                    name=f"{self.name}_{ifo}_state",
                    source_pad_names=("state",),
                    rate=self.state_sample_rate,
                    start=self.start,
                    end=seg_end,
                    segments=state_segments,
                    values=state_values,
                )

                gate = add_state_vector_gating(
                    composed=self,
                    strain_source=noise_source,
                    state_source=state_source,
                    ifo=ifo,
                    bit_mask=self.state_vector_on_dict[ifo],
                    strain_pad=full_channel,
                    state_pad="state",
                    output_pad=full_channel,
                )

                # Add latency tracking if configured
                self._add_latency_tracking(ifo, gate, full_channel)

                if self.verbose:
                    print(
                        f"Added state vector gating for {ifo} with mask "
                        f"{self.state_vector_on_dict[ifo]}"
                    )
        else:
            # No gating - just expose noise source directly
            self.insert(noise_source)

            # Add latency tracking for each IFO
            for ifo in self.ifos:
                full_channel = full_channel_dict[ifo]
                self._add_latency_tracking(ifo, noise_source, full_channel)

Frame Sources

sgnligo.sources.datasource_v2.sources.frames

Frame file composed source classes.

These sources read gravitational wave data from GWF frame files, the standard format for LIGO/Virgo data.

Example

source = FramesComposedSource( ... name="data", ... ifos=["H1", "L1"], ... frame_cache="/path/to/frames.cache", ... channel_dict={"H1": "GDS-CALIB_STRAIN", "L1": "GDS-CALIB_STRAIN"}, ... start=1000000000, ... end=1000000100, ... ) pipeline.connect(source, sink)

FramesComposedSource dataclass

Bases: ComposedSourceBase, ChannelOptionsMixin, FrameCacheOptionsMixin, GPSOptionsMixin, SegmentsOptionsMixin, StateVectorOptionsMixin, InjectionOptionsMixin, VerboseOptionsMixin


              flowchart TD
              sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource[FramesComposedSource]
              sgnligo.sources.composed_base.ComposedSourceBase[ComposedSourceBase]
              sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin[ChannelOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.FrameCacheOptionsMixin[FrameCacheOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsMixin[GPSOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin[SegmentsOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin[StateVectorOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.InjectionOptionsMixin[InjectionOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin[VerboseOptionsMixin]

                              sgnligo.sources.composed_base.ComposedSourceBase --> sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin --> sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.FrameCacheOptionsMixin --> sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsMixin --> sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin --> sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin --> sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.InjectionOptionsMixin --> sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin --> sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource
                


              click sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource href "" "sgnligo.sources.datasource_v2.sources.frames.FramesComposedSource"
              click sgnligo.sources.composed_base.ComposedSourceBase href "" "sgnligo.sources.composed_base.ComposedSourceBase"
              click sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.FrameCacheOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.FrameCacheOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.SegmentsOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.InjectionOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.InjectionOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin"

Frame file source for offline analysis.

Reads strain data from GWF frame files specified in a LAL cache file. Supports optional noiseless injections, segment-based gating, and state vector gating.

Fields inherited from mixins

ifos: List of detector prefixes (from ChannelOptionsMixin) channel_dict: Dict mapping IFO to channel name (from ChannelOptionsMixin) frame_cache: Path to LAL cache file (from FrameCacheOptionsMixin) start: GPS start time (from GPSOptionsMixin) end: GPS end time (from GPSOptionsMixin) segments_file: Path to LIGO XML segments file (from SegmentsOptionsMixin) segments_name: Segment name in XML (from SegmentsOptionsMixin) state_channel_dict: Dict mapping IFO to state vector channel (from StateVectorOptionsMixin) state_vector_on_dict: Dict mapping IFO to bitmask (from StateVectorOptionsMixin) state_segments_file: Path to state segments file (from StateVectorOptionsMixin) state_sample_rate: State vector sample rate (from StateVectorOptionsMixin) noiseless_inj_frame_cache: Injection frame cache (from InjectionOptionsMixin) noiseless_inj_channel_dict: Injection channels (from InjectionOptionsMixin) verbose: Enable verbose output (from VerboseOptionsMixin)

Example

source = FramesComposedSource( ... name="data", ... ifos=["H1"], ... frame_cache="/path/to/frames.cache", ... channel_dict={"H1": "GDS-CALIB_STRAIN"}, ... start=1000000000, ... end=1000000100, ... ) pipeline.connect(source, sink)

Source code in sgnligo/sources/datasource_v2/sources/frames.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
@register_composed_source
@dataclass(kw_only=True)
class FramesComposedSource(
    ComposedSourceBase,
    ChannelOptionsMixin,
    FrameCacheOptionsMixin,
    GPSOptionsMixin,
    SegmentsOptionsMixin,
    StateVectorOptionsMixin,
    InjectionOptionsMixin,
    VerboseOptionsMixin,
):
    """Frame file source for offline analysis.

    Reads strain data from GWF frame files specified in a LAL cache file.
    Supports optional noiseless injections, segment-based gating, and
    state vector gating.

    Fields inherited from mixins:
        ifos: List of detector prefixes (from ChannelOptionsMixin)
        channel_dict: Dict mapping IFO to channel name (from ChannelOptionsMixin)
        frame_cache: Path to LAL cache file (from FrameCacheOptionsMixin)
        start: GPS start time (from GPSOptionsMixin)
        end: GPS end time (from GPSOptionsMixin)
        segments_file: Path to LIGO XML segments file (from SegmentsOptionsMixin)
        segments_name: Segment name in XML (from SegmentsOptionsMixin)
        state_channel_dict: Dict mapping IFO to state vector channel
            (from StateVectorOptionsMixin)
        state_vector_on_dict: Dict mapping IFO to bitmask
            (from StateVectorOptionsMixin)
        state_segments_file: Path to state segments file
            (from StateVectorOptionsMixin)
        state_sample_rate: State vector sample rate
            (from StateVectorOptionsMixin)
        noiseless_inj_frame_cache: Injection frame cache (from InjectionOptionsMixin)
        noiseless_inj_channel_dict: Injection channels (from InjectionOptionsMixin)
        verbose: Enable verbose output (from VerboseOptionsMixin)

    Example:
        >>> source = FramesComposedSource(
        ...     name="data",
        ...     ifos=["H1"],
        ...     frame_cache="/path/to/frames.cache",
        ...     channel_dict={"H1": "GDS-CALIB_STRAIN"},
        ...     start=1000000000,
        ...     end=1000000100,
        ... )
        >>> pipeline.connect(source, sink)
    """

    # Class metadata
    source_type: ClassVar[str] = "frames"
    description: ClassVar[str] = "Read from GWF frame files"

    def _validate(self) -> None:
        """Validate parameters."""
        if self.start is None or self.end is None:
            raise ValueError("start and end are required for FramesComposedSource")
        if self.start >= self.end:
            raise ValueError("start must be less than end")

        # Validate frame cache
        if not os.path.exists(self.frame_cache):
            raise ValueError(f"Frame cache file does not exist: {self.frame_cache}")

        # Validate channel_dict
        if set(self.channel_dict.keys()) != set(self.ifos):
            raise ValueError("channel_dict keys must match ifos")

        # Validate segments options
        if self.segments_file is not None:
            if self.segments_name is None:
                raise ValueError("Must specify segments_name when segments_file is set")
            if not os.path.exists(self.segments_file):
                raise ValueError(f"Segments file does not exist: {self.segments_file}")
            # Every configured channel must have segment definitions in the
            # XML, but the segments file may legitimately declare additional
            # IFOs that this job does not analyze. Offline DAGs run per-IFO
            # jobs (e.g. the reference-PSD layer groups by IFO) against a
            # single shared segments file covering all detectors, so requiring
            # an exact set match would break those jobs. Extra IFOs in the
            # XML take no part in data handling — _build() only iterates
            # self.ifos, and _load_segments() drops detectors not configured
            # for this job — but they are surfaced via all_analysis_ifos so
            # every job reports the same analysis-wide instrument set.
            #
            # Setting require_exact_segment_ifos restores the original strict
            # behavior: the XML must declare exactly the configured IFO set.
            self._load_segments()  # populates self._segments_xml_ifos
            # Full set the XML declares, before _load_segments() filters it.
            xml_ifos = set(getattr(self, "_segments_xml_ifos", set()))
            ifo_set = set(self.ifos)
            missing_in_xml = sorted(ifo_set - xml_ifos)
            extra_in_xml = sorted(xml_ifos - ifo_set)
            if self.require_exact_segment_ifos and (missing_in_xml or extra_in_xml):
                parts = []
                if missing_in_xml:
                    parts.append(
                        f"channels include {missing_in_xml} but the segments "
                        f"file does not"
                    )
                if extra_in_xml:
                    parts.append(
                        f"segments file declares {extra_in_xml} but no "
                        f"channel is configured for them"
                    )
                raise ValueError(
                    "channels and segments file disagree on the analysis "
                    "IFO set: " + "; ".join(parts)
                )
            if missing_in_xml:
                raise ValueError(
                    "segments file is missing segment definitions for "
                    f"configured channels {missing_in_xml}"
                )

        # Validate state vector options (optional for frames)
        if self.state_channel_dict is not None or self.state_vector_on_dict is not None:
            if self.state_channel_dict is None:
                raise ValueError(
                    "Must specify state_channel_dict when state_vector_on_dict is set"
                )
            if self.state_vector_on_dict is None:
                raise ValueError(
                    "Must specify state_vector_on_dict when state_channel_dict is set"
                )
            if set(self.state_channel_dict.keys()) != set(self.ifos):
                raise ValueError("state_channel_dict keys must match ifos")
            if set(self.state_vector_on_dict.keys()) != set(self.ifos):
                raise ValueError("state_vector_on_dict keys must match ifos")

        # Validate injection options
        if self.noiseless_inj_frame_cache is not None:
            if not os.path.exists(self.noiseless_inj_frame_cache):
                raise ValueError(
                    f"Injection frame cache does not exist: "
                    f"{self.noiseless_inj_frame_cache}"
                )
            if self.noiseless_inj_channel_dict is None:
                raise ValueError(
                    "Must specify noiseless_inj_channel_dict when "
                    "noiseless_inj_frame_cache is set"
                )

    def _load_segments(self) -> Optional[Dict[str, List]]:
        """Load and process segments from XML file.

        Memoized to avoid re-parsing the XML on each call (_validate,
        _build, and all_analysis_ifos all consult this).
        """
        if self.segments_file is None or self.segments_name is None:
            return None

        cached = getattr(self, "_segments_dict_memo", None)
        if cached is not None:
            return cached

        loaded_segments = ligolw_segments.segmenttable_get_by_name(
            ligolw_utils.load_filename(
                self.segments_file,
                contenthandler=ligolw_segments.LIGOLWContentHandler,
            ),
            self.segments_name,
        ).coalesce()

        # Record the full set of IFOs the XML declares, before filtering to
        # self.ifos below, so _validate() can enforce exact-match coverage
        # when require_exact_segment_ifos is set.
        self._segments_xml_ifos = set(loaded_segments.keys())

        # Clip to requested time range
        seg = segments.segment(LIGOTimeGPS(self.start), LIGOTimeGPS(self.end))
        clipped_segments = segments.segmentlistdict(
            (ifo, seglist & segments.segmentlist([seg]))
            for ifo, seglist in loaded_segments.items()
        )

        # Convert to nanoseconds, keeping only IFOs this job analyzes. The
        # segments file may cover more detectors than self.ifos (offline DAGs
        # share one file across per-IFO jobs); extras are dropped here so no
        # gating/build logic acts on them, but the full XML set is kept in
        # _segments_xml_ifos for all_analysis_ifos and _validate.
        segments_dict = {}
        for ifo, segs in clipped_segments.items():
            if ifo not in self.ifos:
                continue
            segments_dict[ifo] = [segments.segment(s[0].ns(), s[1].ns()) for s in segs]
        self._segments_dict_memo = segments_dict
        return segments_dict

    @property
    def all_analysis_ifos(self) -> List[str]:
        """IFOs declared in the segments file, or ``self.ifos`` if none.

        This is the *analysis-wide* IFO set, not this job's: with a
        ``segments_file`` configured, it returns every detector the XML
        declares, even ones this job does not analyze (matching v1's
        ``DataSourceInfo.all_analysis_ifos``). Offline DAGs run per-combo
        filter jobs against one shared segments file, and downstream
        likelihood-ratio marginalization requires every job to stamp the
        same instrument set into its output regardless of which subset it
        analyzed. The per-job segments (filtered to ``self.ifos``) live in
        ``_load_segments()``; only this property exposes the full set. The
        property is kept as a distinct concept because the legacy
        ``DataSourceInfo.all_analysis_ifos`` was intended to come from a
        separate source (e.g., a time-slide file in gstlal) and may be
        extended that way later.
        """
        if self._load_segments() is None:
            return list(self.ifos)
        return sorted(self._segments_xml_ifos)

    def _build(self) -> None:
        """Build the frame file source."""
        segments_dict = self._load_segments()

        # Determine sample rate from first frame reader (will be set after creation)
        sample_rate = None

        for ifo in self.ifos:
            channel_name = f"{ifo}:{self.channel_dict[ifo]}"

            # Create main frame reader
            frame_reader = FrameReader(
                name=f"{self.name}_{ifo}_frames",
                framecache=self.frame_cache,
                channel_names=[channel_name],
                instrument=ifo,
                start=self.start,
                end=self.end,
            )

            # Get sample rate from first frame reader
            if sample_rate is None:
                sample_rate = next(iter(frame_reader.rates.values()))

            # Track the current output element and pad for this IFO
            current_source: TSSource | TSTransform = frame_reader
            current_pad = channel_name

            # Add injection if configured
            if self.noiseless_inj_frame_cache and self.noiseless_inj_channel_dict:
                if ifo in self.noiseless_inj_channel_dict:
                    inj_channel = f"{ifo}:{self.noiseless_inj_channel_dict[ifo]}"

                    inj_reader = FrameReader(
                        name=f"{self.name}_{ifo}_inj",
                        framecache=self.noiseless_inj_frame_cache,
                        channel_names=[inj_channel],
                        instrument=ifo,
                        start=self.start,
                        end=self.end,
                    )

                    # Add frames together
                    adder = Adder(
                        name=f"{self.name}_{ifo}_add",
                        sink_pad_names=("frame", "inj"),
                        source_pad_names=(channel_name,),
                    )

                    self.connect(
                        frame_reader,
                        adder,
                        link_map={"frame": channel_name},
                    )
                    self.connect(
                        inj_reader,
                        adder,
                        link_map={"inj": inj_channel},
                    )

                    current_source = adder
                    current_pad = channel_name

                    if self.verbose:
                        print(f"Added injection for {ifo} from {inj_channel}")
            else:
                # No injection - just insert the frame reader
                self.insert(frame_reader)

            # Add state vector gating if configured.
            # Reads the state vector channel from the same frame files and
            # applies BitMask + Gate to gate strain based on data quality.
            if (
                self.state_channel_dict is not None
                and self.state_vector_on_dict is not None
            ):
                state_channel_name = f"{ifo}:{self.state_channel_dict[ifo]}"

                state_reader = FrameReader(
                    name=f"{self.name}_{ifo}_state",
                    framecache=self.frame_cache,
                    channel_names=[state_channel_name],
                    instrument=ifo,
                    start=self.start,
                    end=self.end,
                )

                gate = add_state_vector_gating(
                    composed=self,
                    strain_source=current_source,
                    state_source=state_reader,
                    ifo=ifo,
                    bit_mask=self.state_vector_on_dict[ifo],
                    strain_pad=current_pad,
                    state_pad=state_channel_name,
                    output_pad=channel_name,
                )

                current_source = gate
                current_pad = channel_name

                if self.verbose:
                    print(
                        f"Added state vector gating for {ifo} with mask "
                        f"{self.state_vector_on_dict[ifo]}"
                    )

            # Add segment gating if configured.
            # Always gate when the segments file declares this IFO, even
            # if the segment list is empty after clipping to the GPS window.
            # An empty list means "no valid data" — SegmentSource produces
            # all-gap control and the Gate gates everything.
            if segments_dict is not None and ifo in segments_dict:
                ifo_segments = segments_dict[ifo]

                seg_source = SegmentSource(
                    name=f"{self.name}_{ifo}_seg",
                    source_pad_names=("control",),
                    rate=sample_rate,
                    start=self.start,
                    end=self.end,
                    segments=ifo_segments,  # type: ignore[arg-type]
                )

                gate = Gate(
                    name=f"{self.name}_{ifo}_gate",
                    sink_pad_names=("strain", "control"),
                    control="control",
                    source_pad_names=(channel_name,),
                )

                self.connect(
                    current_source,
                    gate,
                    link_map={"strain": current_pad},
                )
                self.connect(
                    seg_source,
                    gate,
                    link_map={"control": "control"},
                )

                current_source = gate
                current_pad = channel_name

                if self.verbose:
                    if ifo_segments:
                        print(f"Added segment gating for {ifo}")
                    else:
                        print(
                            f"Added segment gating for {ifo} "
                            f"(no valid segments — all data gated)"
                        )

            # Add latency tracking if configured
            self._add_latency_tracking(ifo, current_source, current_pad)

all_analysis_ifos property

IFOs declared in the segments file, or self.ifos if none.

This is the analysis-wide IFO set, not this job's: with a segments_file configured, it returns every detector the XML declares, even ones this job does not analyze (matching v1's DataSourceInfo.all_analysis_ifos). Offline DAGs run per-combo filter jobs against one shared segments file, and downstream likelihood-ratio marginalization requires every job to stamp the same instrument set into its output regardless of which subset it analyzed. The per-job segments (filtered to self.ifos) live in _load_segments(); only this property exposes the full set. The property is kept as a distinct concept because the legacy DataSourceInfo.all_analysis_ifos was intended to come from a separate source (e.g., a time-slide file in gstlal) and may be extended that way later.

DevShm Sources

sgnligo.sources.datasource_v2.sources.devshm

Shared memory (devshm) composed source classes.

These sources read low-latency data from shared memory for online gravitational wave analysis.

Example

source = DevShmComposedSource( ... name="low_latency", ... ifos=["H1"], ... channel_dict={"H1": "GDS-CALIB_STRAIN"}, ... shared_memory_dict={"H1": "/dev/shm/kafka/H1_O4Replay"}, ... state_channel_dict={"H1": "GDS-CALIB_STATE_VECTOR"}, ... state_vector_on_dict={"H1": 3}, ... ) pipeline.connect(source, sink)

DevShmComposedSource dataclass

Bases: ComposedSourceBase, ChannelOptionsMixin, DevShmOptionsMixin, QueueTimeoutOptionsMixin, StateVectorOptionsMixin, VerboseOptionsMixin


              flowchart TD
              sgnligo.sources.datasource_v2.sources.devshm.DevShmComposedSource[DevShmComposedSource]
              sgnligo.sources.composed_base.ComposedSourceBase[ComposedSourceBase]
              sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin[ChannelOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.DevShmOptionsMixin[DevShmOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.QueueTimeoutOptionsMixin[QueueTimeoutOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin[StateVectorOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin[VerboseOptionsMixin]

                              sgnligo.sources.composed_base.ComposedSourceBase --> sgnligo.sources.datasource_v2.sources.devshm.DevShmComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin --> sgnligo.sources.datasource_v2.sources.devshm.DevShmComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.DevShmOptionsMixin --> sgnligo.sources.datasource_v2.sources.devshm.DevShmComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.QueueTimeoutOptionsMixin --> sgnligo.sources.datasource_v2.sources.devshm.DevShmComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin --> sgnligo.sources.datasource_v2.sources.devshm.DevShmComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin --> sgnligo.sources.datasource_v2.sources.devshm.DevShmComposedSource
                


              click sgnligo.sources.datasource_v2.sources.devshm.DevShmComposedSource href "" "sgnligo.sources.datasource_v2.sources.devshm.DevShmComposedSource"
              click sgnligo.sources.composed_base.ComposedSourceBase href "" "sgnligo.sources.composed_base.ComposedSourceBase"
              click sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.DevShmOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.DevShmOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.QueueTimeoutOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.QueueTimeoutOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin"

Shared memory source with state vector gating.

Reads low-latency strain data from shared memory and applies state vector gating to ensure only valid data is processed.

Fields inherited from mixins

ifos: List of detector prefixes (from ChannelOptionsMixin) channel_dict: Dict mapping IFO to channel name (from ChannelOptionsMixin) shared_memory_dict: Dict mapping IFO to shm path (from DevShmOptionsMixin) discont_wait_time: Discontinuity wait time (from DevShmOptionsMixin) queue_timeout: Queue timeout (from QueueTimeoutOptionsMixin) state_channel_dict: Dict mapping IFO to state vector channel (from StateVectorOptionsMixin) state_vector_on_dict: Dict mapping IFO to bitmask (from StateVectorOptionsMixin) state_segments_file: Path to state segments file (from StateVectorOptionsMixin) state_sample_rate: State vector sample rate (from StateVectorOptionsMixin) verbose: Enable verbose output (from VerboseOptionsMixin)

Note

state_channel_dict and state_vector_on_dict are required for DevShm sources (validation will fail if not provided).

Example

source = DevShmComposedSource( ... name="low_latency", ... ifos=["H1"], ... channel_dict={"H1": "GDS-CALIB_STRAIN"}, ... shared_memory_dict={"H1": "/dev/shm/kafka/H1_O4Replay"}, ... state_channel_dict={"H1": "GDS-CALIB_STATE_VECTOR"}, ... state_vector_on_dict={"H1": 3}, ... ) pipeline.connect(source, sink)

Source code in sgnligo/sources/datasource_v2/sources/devshm.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
@register_composed_source
@dataclass(kw_only=True)
class DevShmComposedSource(
    ComposedSourceBase,
    ChannelOptionsMixin,
    DevShmOptionsMixin,
    QueueTimeoutOptionsMixin,
    StateVectorOptionsMixin,
    VerboseOptionsMixin,
):
    """Shared memory source with state vector gating.

    Reads low-latency strain data from shared memory and applies state
    vector gating to ensure only valid data is processed.

    Fields inherited from mixins:
        ifos: List of detector prefixes (from ChannelOptionsMixin)
        channel_dict: Dict mapping IFO to channel name (from ChannelOptionsMixin)
        shared_memory_dict: Dict mapping IFO to shm path (from DevShmOptionsMixin)
        discont_wait_time: Discontinuity wait time (from DevShmOptionsMixin)
        queue_timeout: Queue timeout (from QueueTimeoutOptionsMixin)
        state_channel_dict: Dict mapping IFO to state vector channel
            (from StateVectorOptionsMixin)
        state_vector_on_dict: Dict mapping IFO to bitmask
            (from StateVectorOptionsMixin)
        state_segments_file: Path to state segments file (from StateVectorOptionsMixin)
        state_sample_rate: State vector sample rate (from StateVectorOptionsMixin)
        verbose: Enable verbose output (from VerboseOptionsMixin)

    Note:
        state_channel_dict and state_vector_on_dict are required for DevShm sources
        (validation will fail if not provided).

    Example:
        >>> source = DevShmComposedSource(
        ...     name="low_latency",
        ...     ifos=["H1"],
        ...     channel_dict={"H1": "GDS-CALIB_STRAIN"},
        ...     shared_memory_dict={"H1": "/dev/shm/kafka/H1_O4Replay"},
        ...     state_channel_dict={"H1": "GDS-CALIB_STATE_VECTOR"},
        ...     state_vector_on_dict={"H1": 3},
        ... )
        >>> pipeline.connect(source, sink)
    """

    # Class metadata
    source_type: ClassVar[str] = "devshm"
    description: ClassVar[str] = "Read from shared memory"

    def _validate(self) -> None:
        """Validate parameters."""
        ifos_set = set(self.ifos)

        # Validate channel_dict
        if set(self.channel_dict.keys()) != ifos_set:
            raise ValueError("channel_dict keys must match ifos")

        # Validate shared_memory_dict
        if set(self.shared_memory_dict.keys()) != ifos_set:
            raise ValueError("shared_memory_dict keys must match ifos")

        # Validate state_channel_dict (required for devshm)
        if self.state_channel_dict is None:
            raise ValueError("state_channel_dict is required for DevShm sources")
        if set(self.state_channel_dict.keys()) != ifos_set:
            raise ValueError("state_channel_dict keys must match ifos")

        # Validate state_vector_on_dict (required for devshm)
        if self.state_vector_on_dict is None:
            raise ValueError("state_vector_on_dict is required for DevShm sources")
        if set(self.state_vector_on_dict.keys()) != ifos_set:
            raise ValueError("state_vector_on_dict keys must match ifos")

    def _build(self) -> None:
        """Build the shared memory source with state vector gating."""
        # These are validated as required in _validate()
        assert self.state_channel_dict is not None
        assert self.state_vector_on_dict is not None

        # Build channel names for DevShmSource
        # DevShmSource expects: {ifo: [strain_channel, state_channel]}
        channel_names = {}
        for ifo in self.ifos:
            strain_channel = f"{ifo}:{self.channel_dict[ifo]}"
            state_channel = f"{ifo}:{self.state_channel_dict[ifo]}"
            channel_names[ifo] = [strain_channel, state_channel]

        # Create the shared memory source
        devshm = DevShmSource(
            name=f"{self.name}_devshm",
            channel_names=channel_names,
            shared_memory_dirs=self.shared_memory_dict,
            discont_wait_time=self.discont_wait_time,
            queue_timeout=self.queue_timeout,
            verbose=self.verbose,
        )

        # Add state vector gating for each IFO
        for ifo in self.ifos:
            strain_channel = f"{ifo}:{self.channel_dict[ifo]}"
            state_channel = f"{ifo}:{self.state_channel_dict[ifo]}"

            gate = add_state_vector_gating(
                composed=self,
                strain_source=devshm,
                state_source=devshm,
                ifo=ifo,
                bit_mask=self.state_vector_on_dict[ifo],
                strain_pad=strain_channel,
                state_pad=state_channel,
                output_pad=strain_channel,
            )

            # Add latency tracking if configured
            self._add_latency_tracking(ifo, gate, strain_channel)

            if self.verbose:
                print(
                    f"Added state vector gating for {ifo} with mask "
                    f"{self.state_vector_on_dict[ifo]}"
                )

Arrakis Sources

sgnligo.sources.datasource_v2.sources.arrakis

Arrakis composed source classes.

These sources read streaming data via Arrakis for online gravitational wave analysis.

Example

source = ArrakisComposedSource( ... name="kafka_data", ... ifos=["H1", "L1"], ... channel_dict={"H1": "GDS-CALIB_STRAIN", "L1": "GDS-CALIB_STRAIN"}, ... ) pipeline.connect(source, sink)

ArrakisComposedSource dataclass

Bases: ComposedSourceBase, ChannelOptionsMixin, GPSOptionsFlexibleMixin, QueueTimeoutOptionsMixin, ReplayOptionsMixin, StateVectorOptionsMixin, VerboseOptionsMixin


              flowchart TD
              sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource[ArrakisComposedSource]
              sgnligo.sources.composed_base.ComposedSourceBase[ComposedSourceBase]
              sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin[ChannelOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin[GPSOptionsFlexibleMixin]
              sgnligo.sources.datasource_v2.cli_mixins.QueueTimeoutOptionsMixin[QueueTimeoutOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.ReplayOptionsMixin[ReplayOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin[StateVectorOptionsMixin]
              sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin[VerboseOptionsMixin]

                              sgnligo.sources.composed_base.ComposedSourceBase --> sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin --> sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin --> sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.QueueTimeoutOptionsMixin --> sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.ReplayOptionsMixin --> sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin --> sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource
                
                sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin --> sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource
                


              click sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource href "" "sgnligo.sources.datasource_v2.sources.arrakis.ArrakisComposedSource"
              click sgnligo.sources.composed_base.ComposedSourceBase href "" "sgnligo.sources.composed_base.ComposedSourceBase"
              click sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.ChannelOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.GPSOptionsFlexibleMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.QueueTimeoutOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.QueueTimeoutOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.ReplayOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.ReplayOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.StateVectorOptionsMixin"
              click sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin href "" "sgnligo.sources.datasource_v2.cli_mixins.VerboseOptionsMixin"

Arrakis source for streaming data.

Reads streaming gravitational wave data from topics. Optionally supports state vector gating.

This source defaults to real-time mode. GPS times are optional.

Fields inherited from mixins

ifos: List of detector prefixes (from ChannelOptionsMixin) channel_dict: Dict mapping IFO to channel name (from ChannelOptionsMixin) start: GPS start time (optional, from GPSOptionsFlexibleMixin) end: GPS end time (optional, from GPSOptionsFlexibleMixin) duration: Duration in seconds (optional, from GPSOptionsFlexibleMixin) real_time: Enable real-time mode (default: True, from GPSOptionsFlexibleMixin) queue_timeout: Queue timeout (from QueueTimeoutOptionsMixin) replay_id: Server-registered replay identifier (from ReplayOptionsMixin) state_channel_dict: State channel dict (from StateVectorOptionsMixin) state_vector_on_dict: Bitmask dict (from StateVectorOptionsMixin) state_segments_file: State segments file (from StateVectorOptionsMixin) state_sample_rate: State vector sample rate (from StateVectorOptionsMixin) verbose: Enable verbose output (from VerboseOptionsMixin)

Example

source = ArrakisComposedSource( ... name="kafka_data", ... ifos=["H1"], ... channel_dict={"H1": "GDS-CALIB_STRAIN"}, ... ) pipeline.connect(source, sink)

Source code in sgnligo/sources/datasource_v2/sources/arrakis.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@register_composed_source
@dataclass(kw_only=True)
class ArrakisComposedSource(
    ComposedSourceBase,
    ChannelOptionsMixin,
    GPSOptionsFlexibleMixin,
    QueueTimeoutOptionsMixin,
    ReplayOptionsMixin,
    StateVectorOptionsMixin,
    VerboseOptionsMixin,
):
    """Arrakis source for streaming data.

    Reads streaming gravitational wave data from topics.
    Optionally supports state vector gating.

    This source defaults to real-time mode. GPS times are optional.

    Fields inherited from mixins:
        ifos: List of detector prefixes (from ChannelOptionsMixin)
        channel_dict: Dict mapping IFO to channel name (from ChannelOptionsMixin)
        start: GPS start time (optional, from GPSOptionsFlexibleMixin)
        end: GPS end time (optional, from GPSOptionsFlexibleMixin)
        duration: Duration in seconds (optional, from GPSOptionsFlexibleMixin)
        real_time: Enable real-time mode (default: True, from GPSOptionsFlexibleMixin)
        queue_timeout: Queue timeout (from QueueTimeoutOptionsMixin)
        replay_id: Server-registered replay identifier (from ReplayOptionsMixin)
        state_channel_dict: State channel dict (from StateVectorOptionsMixin)
        state_vector_on_dict: Bitmask dict (from StateVectorOptionsMixin)
        state_segments_file: State segments file (from StateVectorOptionsMixin)
        state_sample_rate: State vector sample rate (from StateVectorOptionsMixin)
        verbose: Enable verbose output (from VerboseOptionsMixin)

    Example:
        >>> source = ArrakisComposedSource(
        ...     name="kafka_data",
        ...     ifos=["H1"],
        ...     channel_dict={"H1": "GDS-CALIB_STRAIN"},
        ... )
        >>> pipeline.connect(source, sink)
    """

    # Class metadata
    source_type: ClassVar[str] = "arrakis"
    description: ClassVar[str] = "Read from Arrakis"

    # Override real_time default to True for streaming source
    real_time: bool = True

    def _validate(self) -> None:
        """Validate parameters."""
        ifos_set = set(self.ifos)

        # Validate channel_dict
        if set(self.channel_dict.keys()) != ifos_set:
            raise ValueError("channel_dict keys must match ifos")

        # Validate time range if both provided
        if self.start is not None and self.end is not None and self.start >= self.end:
            raise ValueError("start must be less than end")

        # Validate state vector options
        if self.state_channel_dict is not None:
            if set(self.state_channel_dict.keys()) != ifos_set:
                raise ValueError("state_channel_dict keys must match ifos")
            if self.state_vector_on_dict is None:
                raise ValueError(
                    "Must specify state_vector_on_dict when state_channel_dict is set"
                )

        if self.state_vector_on_dict is not None:
            if set(self.state_vector_on_dict.keys()) != ifos_set:
                raise ValueError("state_vector_on_dict keys must match ifos")
            if self.state_channel_dict is None:
                raise ValueError(
                    "Must specify state_channel_dict when state_vector_on_dict is set"
                )

    def _build(self) -> None:
        """Build the Arrakis source."""
        # Check if state vector gating is enabled
        use_state_vector = (
            self.state_channel_dict is not None
            and self.state_vector_on_dict is not None
        )

        # Build channel names list for ArrakisSource
        channel_names = []
        for ifo in self.ifos:
            strain_channel = f"{ifo}:{self.channel_dict[ifo]}"
            channel_names.append(strain_channel)

            if use_state_vector:
                assert self.state_channel_dict is not None  # for type checker
                state_channel = f"{ifo}:{self.state_channel_dict[ifo]}"
                channel_names.append(state_channel)

        # Calculate duration from end time (handles duration field too)
        end_time = self._get_computed_end()
        duration = None
        if self.start is not None and end_time is not None:
            duration = end_time - self.start

        # Create the Arrakis source
        arrakis = ArrakisSource(
            name=f"{self.name}_arrakis",
            channels=channel_names,
            replay_id=self.replay_id,
            start=self.start,
            duration=duration,
            in_queue_timeout=int(self.queue_timeout),
        )

        if use_state_vector:
            # Add state vector gating for each IFO
            assert self.state_channel_dict is not None  # for type checker
            assert self.state_vector_on_dict is not None  # for type checker
            for ifo in self.ifos:
                strain_channel = f"{ifo}:{self.channel_dict[ifo]}"
                state_channel = f"{ifo}:{self.state_channel_dict[ifo]}"

                gate = add_state_vector_gating(
                    composed=self,
                    strain_source=arrakis,
                    state_source=arrakis,
                    ifo=ifo,
                    bit_mask=self.state_vector_on_dict[ifo],
                    strain_pad=strain_channel,
                    state_pad=state_channel,
                    output_pad=strain_channel,
                )

                # Add latency tracking if configured
                self._add_latency_tracking(ifo, gate, strain_channel)

                if self.verbose:
                    print(
                        f"Added state vector gating for {ifo} with mask "
                        f"{self.state_vector_on_dict[ifo]}"
                    )
        else:
            # No gating - just expose Arrakis source directly
            self.insert(arrakis)

            # Add latency tracking for each IFO
            for ifo in self.ifos:
                strain_channel = f"{ifo}:{self.channel_dict[ifo]}"
                self._add_latency_tracking(ifo, arrakis, strain_channel)

Base Classes

sgnligo.sources.composed_base

Base class for composed source elements.

This module provides a base class for creating composed source elements that combine multiple internal elements into a single source. Subclasses declare parameters as dataclass fields and override _build() to wire internal elements using self.insert() and self.connect(). The resulting object IS a TS source element — pass it directly to pipeline.connect().

Example

from dataclasses import dataclass from typing import ClassVar, List from sgnts.sources import FakeSeriesSource from sgnligo.sources.composed_base import ComposedSourceBase

@dataclass(kw_only=True) ... class WhiteSource(ComposedSourceBase): ... source_type: ClassVar[str] = "white" ... description: ClassVar[str] = "Gaussian white noise" ... ... ifos: List[str] ... sample_rate: int ... start: float ... end: float ... ... def build(self): ... for ifo in self.ifos: ... self.insert(FakeSeriesSource( ... name=f"{self.name}{ifo}", ... source_pad_names=(f"{ifo}:STRAIN",), ... rate=self.sample_rate, ... start=self.start, ... end=self.end, ... signal_type="white", ... ))

source = WhiteSource( ... name="noise", ifos=["H1", "L1"], ... sample_rate=4096, start=1000, end=1010, ... ) pipeline.connect(source, sink)

ComposedSourceBase dataclass

Bases: TSComposedSourceElement


              flowchart TD
              sgnligo.sources.composed_base.ComposedSourceBase[ComposedSourceBase]

              

              click sgnligo.sources.composed_base.ComposedSourceBase href "" "sgnligo.sources.composed_base.ComposedSourceBase"

Base class for composed source elements.

Subclasses declare their parameters as dataclass fields and override _build() to wire internal elements using self.insert() and self.connect(). __post_init__ (inherited from TSComposedSourceElement) calls _validate() then _build() then sets up boundary pads.

This class adds to the bare TSComposedSourceElement API:

  • source_type / description class vars for CLI/registry use.
  • latency_interval field and _add_latency_tracking() helper for optional latency-monitoring side outputs.
  • CLI integration (add_cli_arguments / get_cli_arg_names / process_cli_args).
Class Attributes

source_type: String identifier for registry (e.g., "white", "frames"). Leave empty for classes that should not be registered. description: Human-readable description for help text.

Source code in sgnligo/sources/composed_base.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@dataclass(kw_only=True)
class ComposedSourceBase(TSComposedSourceElement):
    """Base class for composed source elements.

    Subclasses declare their parameters as dataclass fields and override
    ``_build()`` to wire internal elements using ``self.insert()`` and
    ``self.connect()``. ``__post_init__`` (inherited from
    ``TSComposedSourceElement``) calls ``_validate()`` then ``_build()``
    then sets up boundary pads.

    This class adds to the bare ``TSComposedSourceElement`` API:

    - ``source_type`` / ``description`` class vars for CLI/registry use.
    - ``latency_interval`` field and ``_add_latency_tracking()`` helper for
      optional latency-monitoring side outputs.
    - CLI integration (``add_cli_arguments`` / ``get_cli_arg_names`` /
      ``process_cli_args``).

    Class Attributes:
        source_type: String identifier for registry (e.g., "white",
            "frames"). Leave empty for classes that should not be
            registered.
        description: Human-readable description for help text.
    """

    # Optional latency tracking (interval in seconds, None = disabled)
    latency_interval: Optional[float] = None

    # Class-level metadata for registry and CLI; subclasses override these
    source_type: ClassVar[str] = ""
    description: ClassVar[str] = ""

    def _add_latency_tracking(
        self,
        ifo: str,
        strain_source_element: TSSource | TSTransform,
        strain_pad_name: str,
    ) -> None:
        """Attach latency tracking to a per-IFO strain output.

        Call during ``_build()`` after creating each strain source. The
        strain pad is wired to a ``Latency`` element for monitoring and
        also marked for external exposure (multilink pattern) so the
        strain can flow out of this composed element as well.

        The latency output appears as an additional source pad named
        ``{ifo}_latency``.

        Args:
            ifo: IFO name (e.g., "H1").
            strain_source_element: Element producing the strain data.
            strain_pad_name: Pad name on ``strain_source_element`` to tap.

        Example:
            def _build(self):
                for ifo in self.ifos:
                    source = FakeSeriesSource(...)
                    self.insert(source)
                    self._add_latency_tracking(ifo, source, ifo)
        """
        if self.latency_interval is None:
            return

        latency = Latency(
            name=f"{self.name}_{ifo}_latency",
            sink_pad_names=("data",),
            source_pad_names=(f"{ifo}_latency",),
            route=f"{ifo}_datasource_latency",
            interval=self.latency_interval,
        )

        # Wire strain → latency; the strain pad is internally consumed here,
        # so we also expose it externally via the multilink mechanism.
        self.connect(
            strain_source_element,
            latency,
            link_map={"data": strain_pad_name},
        )
        self.expose_source_pad(f"{strain_source_element.name}:src:{strain_pad_name}")

    # --- CLI argument support ---

    @classmethod
    def add_cli_arguments(cls, parser: argparse.ArgumentParser) -> None:
        """Add CLI arguments for latency tracking."""
        parser.add_argument(
            "--source-latency-interval",
            type=float,
            metavar="SECONDS",
            default=None,
            help="Enable source latency tracking with specified interval in seconds",
        )

    @classmethod
    def get_cli_arg_names(cls) -> Set[str]:
        """Return set of CLI argument names defined by this class."""
        return {"source_latency_interval"}

    @classmethod
    def process_cli_args(cls, args: argparse.Namespace) -> Dict[str, Any]:
        """Convert CLI args to field values."""
        result: Dict[str, Any] = {}
        source_latency_interval = getattr(args, "source_latency_interval", None)
        if source_latency_interval is not None:
            result["latency_interval"] = source_latency_interval
        return result

add_cli_arguments(parser) classmethod

Add CLI arguments for latency tracking.

Source code in sgnligo/sources/composed_base.py 
137
138
139
140
141
142
143
144
145
146
@classmethod
def add_cli_arguments(cls, parser: argparse.ArgumentParser) -> None:
    """Add CLI arguments for latency tracking."""
    parser.add_argument(
        "--source-latency-interval",
        type=float,
        metavar="SECONDS",
        default=None,
        help="Enable source latency tracking with specified interval in seconds",
    )

get_cli_arg_names() classmethod

Return set of CLI argument names defined by this class.

Source code in sgnligo/sources/composed_base.py 
148
149
150
151
@classmethod
def get_cli_arg_names(cls) -> Set[str]:
    """Return set of CLI argument names defined by this class."""
    return {"source_latency_interval"}

process_cli_args(args) classmethod

Convert CLI args to field values.

Source code in sgnligo/sources/composed_base.py 
153
154
155
156
157
158
159
160
@classmethod
def process_cli_args(cls, args: argparse.Namespace) -> Dict[str, Any]:
    """Convert CLI args to field values."""
    result: Dict[str, Any] = {}
    source_latency_interval = getattr(args, "source_latency_interval", None)
    if source_latency_interval is not None:
        result["latency_interval"] = source_latency_interval
    return result

Utilities

sgnligo.sources.datasource_v2.sources.utils

Utility functions for composed sources.

This module contains reusable building blocks that are shared across multiple composed source classes.

add_state_vector_gating(composed, strain_source, state_source, ifo, bit_mask, strain_pad, state_pad, output_pad)

Wire a BitMask + Gate into composed for state vector gating.

Common pattern for devshm, arrakis, and gwdata-noise sources. Applies a bitmask to the state vector channel, then uses a Gate to control strain data based on the masked state vector.

The pattern is

strain_source[strain_pad] ─────────────────┐ ├─> Gate[output_pad] state_source[state_pad] -> BitMask[state] ─┘

Parameters:

Name Type Description Default
composed ComposedElementMixin

The composed element being built (a ComposedSourceBase or ComposedTransformBase). Call only during _build().

 required
strain_source

Source element providing strain data.

 required
state_source

Source element providing state vector data.

 required
ifo str

Interferometer prefix (e.g., "H1").

 required
bit_mask int

Bitmask to apply to state vector.

 required
strain_pad str

Name of the strain output pad on strain_source.

 required
state_pad str

Name of the state vector output pad on state_source.

 required
output_pad str

Name for the gated output pad.

 required

Returns:

Type Description
Gate

The Gate element for downstream use (e.g., latency tracking).
Source code in sgnligo/sources/datasource_v2/sources/utils.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def add_state_vector_gating(
    composed: ComposedElementMixin,
    strain_source,
    state_source,
    ifo: str,
    bit_mask: int,
    strain_pad: str,
    state_pad: str,
    output_pad: str,
) -> Gate:
    """Wire a BitMask + Gate into ``composed`` for state vector gating.

    Common pattern for devshm, arrakis, and gwdata-noise sources. Applies
    a bitmask to the state vector channel, then uses a Gate to control
    strain data based on the masked state vector.

    The pattern is:
        strain_source[strain_pad] ─────────────────┐
                                                   ├─> Gate[output_pad]
        state_source[state_pad] -> BitMask[state] ─┘

    Args:
        composed: The composed element being built (a ``ComposedSourceBase``
            or ``ComposedTransformBase``). Call only during ``_build()``.
        strain_source: Source element providing strain data.
        state_source: Source element providing state vector data.
        ifo: Interferometer prefix (e.g., "H1").
        bit_mask: Bitmask to apply to state vector.
        strain_pad: Name of the strain output pad on ``strain_source``.
        state_pad: Name of the state vector output pad on ``state_source``.
        output_pad: Name for the gated output pad.

    Returns:
        The Gate element for downstream use (e.g., latency tracking).
    """
    mask = BitMask(
        name=f"{ifo}_Mask",
        sink_pad_names=("state",),
        source_pad_names=("state",),
        bit_mask=bit_mask,
    )

    gate = Gate(
        name=f"{ifo}_Gate",
        sink_pad_names=("strain", "state_vector"),
        control="state_vector",
        source_pad_names=(output_pad,),
    )

    composed.connect(state_source, mask, link_map={"state": state_pad})
    composed.connect(mask, gate, link_map={"state_vector": "state"})
    composed.connect(strain_source, gate, link_map={"strain": strain_pad})

    return gate