Skip to content

sgnligo.base.utils

A module that contains utility functions for SGN-LIGO programs.

from_T050017(url)

Parse a URL in the style of T050017-00.

Source code in sgnligo/base/utils.py 
83
84
85
86
87
88
89
def from_T050017(url):
    """
    Parse a URL in the style of T050017-00.
    """
    filename, _ = os.path.splitext(os.path.basename(url))
    obs, desc, start, dur = filename.split("-")
    return obs, desc, int(start), int(dur)

now()

A convenience function to return the current gps time

Source code in sgnligo/base/utils.py 
17
18
19
20
21
def now():
    """
    A convenience function to return the current gps time
    """
    return gwpy.time.to_gps(datetime.utcnow())

parse_list_to_dict(lst, value_transform=None, sep='=', key_is_range=False, range_sep=':')

A general list to dict argument parsing coercion function

Parameters:

Name Type Description Default
lst list

list, a list of the form ['A=V1', 'B=V2', ...], where "=" only has to match the sep_str argument

 required
value_transform Optional[FunctionType]

Function, default None. An optional transformation function to apply on values of the dictionary

 None
sep str

str, default '=', the separator string between dict keys and values in list elements

 '='
key_is_range bool

bool, default False. If True, the keys of the list are compound and contain range information e.g. "start:stop:remaining,list,of,items"

 False
range_sep str

str, default ':' the separator string for range key information

 ':'

Returns:

Type Description
dict

dict of the form {'A': value_transform('V1'), ...}

Examples:

>>> parse_list_to_dict(["H1=LSC-STRAIN", "H2=SOMETHING-ELSE"])
{'H1': 'LSC-STRAIN', 'H2': 'SOMETHING-ELSE'}

>>> parse_list_to_dict(["0000:0002:H1=LSC_STRAIN_1,L1=LSC_STRAIN_2",
...     "0002:0004:H1=LSC_STRAIN_3,L1=LSC_STRAIN_4",
...     "0004:0006:H1=LSC_STRAIN_5,L1=LSC_STRAIN_6"],
...     key_is_range=True)
{'0000': {'H1': 'LSC_STRAIN_1', 'L1': 'LSC_STRAIN_2'},
        '0001': {'H1': 'LSC_STRAIN_1', 'L1': 'LSC_STRAIN_2'},
        '0002': {'H1': 'LSC_STRAIN_3', 'L1': 'LSC_STRAIN_4'},
        '0003': {'H1': 'LSC_STRAIN_3', 'L1': 'LSC_STRAIN_4'},
        '0004': {'H1': 'LSC_STRAIN_5', 'L1': 'LSC_STRAIN_6'},
        '0005': {'H1': 'LSC_STRAIN_5', 'L1': 'LSC_STRAIN_6'}}
Source code in sgnligo/base/utils.py 
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
def parse_list_to_dict(
    lst: list,
    value_transform: Optional[types.FunctionType] = None,
    sep: str = "=",
    key_is_range: bool = False,
    range_sep: str = ":",
) -> dict:
    """A general list to dict argument parsing coercion function

    Args:
        lst:
            list, a list of the form ['A=V1', 'B=V2', ...], where "=" only has to
            match the sep_str argument
        value_transform:
            Function, default None. An optional transformation function to apply on
            values of the dictionary
        sep:
            str, default '=', the separator string between dict keys and values in list
            elements
        key_is_range:
            bool, default False. If True, the keys of the list are compound and contain
            range information e.g. "start:stop:remaining,list,of,items"
        range_sep:
            str, default ':' the separator string for range key information

    Returns:
        dict of the form {'A': value_transform('V1'), ...}

    Examples:
        >>> parse_list_to_dict(["H1=LSC-STRAIN", "H2=SOMETHING-ELSE"])  # doctest: +SKIP
        {'H1': 'LSC-STRAIN', 'H2': 'SOMETHING-ELSE'}

        >>> parse_list_to_dict(["0000:0002:H1=LSC_STRAIN_1,L1=LSC_STRAIN_2",
        ...     "0002:0004:H1=LSC_STRAIN_3,L1=LSC_STRAIN_4",
        ...     "0004:0006:H1=LSC_STRAIN_5,L1=LSC_STRAIN_6"],
        ...     key_is_range=True)  # doctest: +SKIP
        {'0000': {'H1': 'LSC_STRAIN_1', 'L1': 'LSC_STRAIN_2'},
                '0001': {'H1': 'LSC_STRAIN_1', 'L1': 'LSC_STRAIN_2'},
                '0002': {'H1': 'LSC_STRAIN_3', 'L1': 'LSC_STRAIN_4'},
                '0003': {'H1': 'LSC_STRAIN_3', 'L1': 'LSC_STRAIN_4'},
                '0004': {'H1': 'LSC_STRAIN_5', 'L1': 'LSC_STRAIN_6'},
                '0005': {'H1': 'LSC_STRAIN_5', 'L1': 'LSC_STRAIN_6'}}
    """
    if lst is None:
        return
    coerced = {}
    if key_is_range:
        # This will produce tuples (start, stop, str-to-dict)
        splits = [e.split(range_sep) for e in lst]
        for start, stop, val in splits:
            for i in range(int(start), int(stop)):
                key = str(i).zfill(4)
                coerced[key] = parse_list_to_dict(
                    [v.strip() for v in val.split(",")],
                    value_transform=value_transform,
                    sep=sep,
                    key_is_range=False,
                )
    else:
        if len(lst) == 1 and sep not in lst[0]:  # non-dict entry
            return lst[0]
        coerced = dict([e.split(sep) for e in lst])
        if value_transform is not None:
            for k in coerced:
                coerced[k] = value_transform(coerced[k])
        return coerced
    return coerced

read_segments_and_values_from_file(filename, verbose=False)

Read time segments and associated values from a file.

This function reads a text file defining time segments with associated values, typically used for state vectors or data quality flags. The file format is:

start_gps end_gps value

where times are in GPS seconds and value is an integer (e.g., bitmask value).

Note: This is different from LIGO segment files which define only time intervals. This function reads segment-value pairs for things like state vector patterns.

Parameters:

Name Type Description Default
filename

Path to the segments+values file

 required
verbose

Whether to print verbose output

 False

Returns:

Name Type Description
tuple

(segments, values) where segments are tuples of (start_ns, end_ns) in nanoseconds and values are the corresponding integer values

Examples:

>>> segments, values = read_segments_and_values_from_file("state_data.txt")
>>> # segments = ((1400000000000000000, 1400000016000000000), ...)
>>> # values = (1, 3, 7, ...)
Source code in sgnligo/base/utils.py 
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
69
70
71
72
73
74
75
76
77
78
79
80
def read_segments_and_values_from_file(filename, verbose=False):
    """Read time segments and associated values from a file.

    This function reads a text file defining time segments with associated values,
    typically used for state vectors or data quality flags. The file format is:

        start_gps end_gps value

    where times are in GPS seconds and value is an integer (e.g., bitmask value).

    Note: This is different from LIGO segment files which define only time intervals.
    This function reads segment-value pairs for things like state vector patterns.

    Args:
        filename: Path to the segments+values file
        verbose: Whether to print verbose output

    Returns:
        tuple: (segments, values) where segments are tuples of (start_ns, end_ns)
               in nanoseconds and values are the corresponding integer values

    Examples:
        >>> segments, values = read_segments_and_values_from_file("state_data.txt")
        >>> # segments = ((1400000000000000000, 1400000016000000000), ...)
        >>> # values = (1, 3, 7, ...)
    """
    if verbose:
        print(f"Reading segments and values from {filename}")

    # Read the file
    data = np.loadtxt(filename)

    # Ensure we have 3 columns
    if data.ndim == 1:
        # Single row
        data = data.reshape(1, -1)

    if data.shape[1] != 3:
        raise ValueError(
            f"Segments file must have 3 columns (start end value), got {data.shape[1]}"
        )

    segments = []
    values = []

    for i, (start, end, value) in enumerate(data):
        # Convert times to nanoseconds
        start_ns = int(start * 1e9)
        end_ns = int(end * 1e9)

        segments.append((start_ns, end_ns))
        values.append(int(value))

        if verbose:
            print(f"  Segment {i+1}: {start}s - {end}s, Value: {int(value)}")

    return tuple(segments), tuple(values)

state_vector_on_off_bits(bit)

Format the given bitmask appropriately as an integer

Source code in sgnligo/base/utils.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def state_vector_on_off_bits(bit):
    """
    Format the given bitmask appropriately as an integer
    """
    if isinstance(bit, str):
        if not bit.startswith("0b"):
            bit = "0b" + bit
        bit = int(bit, 2)
    else:
        bit = int(bit)

    return bit