Skip to content

sgnl.sgnlio

SgnlDB dataclass

Bases: StillSuit

Source code in sgnl/sgnlio.py 
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@dataclass
class SgnlDB(stillsuit.StillSuit):
    def __post_init__(self):
        super().__post_init__()

    def segments(self, name="afterhtgate"):
        """
        Retrieve segments with the specified name from the database and organize them
        into a dictionary by interferometer (ifo) combination.
        """
        segments = {}
        for row in self.db.cursor().execute(
            "SELECT * FROM segment WHERE name = ?", (name,)
        ):
            row = dict(row)
            segments.setdefault(row["ifo"], []).append(
                (row["start_time"], row["end_time"])
            )

        # Convert rows to segment list dictionary
        segments = segmentlistdict(
            {
                ifo: segmentlist(segment(*interval) for interval in intervals)
                for ifo, intervals in segments.items()
            }
        )

        # Generate all possible ifo combinations
        ifos = frozenset(segments.keys())
        combos = [
            frozenset(combo)
            for level in range(len(ifos), 0, -1)
            for combo in itertools.combinations(ifos, level)
        ]

        # Create the output dictionary
        out = {
            combo: segments.intersection(combo) - segments.union(ifos - combo)
            for combo in combos
        }

        return segmentlistdict(out)

    def missed_found_by_on_ifos(
        self, far_threshold=1 / 86400 / 365.25, segments_name="afterhtgate"
    ):
        """
        get missed and found instruments by on ifos
        FIXME I am sure this is stupidly slow
        """

        _missed, _found = self.get_missed_found(
            selection_func=lambda r: r["event"]["combined_far"] <= far_threshold
        )
        _segments = self.segments(name=segments_name)

        missed = _MissedByOnIFOs(_segments, _missed, self.schema)
        found = _FoundByOnIFOs(_segments, _found, self.schema)

        return missed, found

    def get_events(self, nanosec_to_sec=False, **kwargs):
        """
        A wrapper function of StillSuit.get_events() with additional
        functionalities
        """

        for event in super(SgnlDB, self).get_events(**kwargs):
            if nanosec_to_sec:
                for trigger in event["trigger"]:
                    trigger["time"] *= 1e-9
                    trigger["epoch_start"] *= 1e-9
                    trigger["epoch_end"] *= 1e-9
                event["event"]["time"] *= 1e-9
            yield event

get_events(nanosec_to_sec=False, **kwargs)

A wrapper function of StillSuit.get_events() with additional functionalities

Source code in sgnl/sgnlio.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_events(self, nanosec_to_sec=False, **kwargs):
    """
    A wrapper function of StillSuit.get_events() with additional
    functionalities
    """

    for event in super(SgnlDB, self).get_events(**kwargs):
        if nanosec_to_sec:
            for trigger in event["trigger"]:
                trigger["time"] *= 1e-9
                trigger["epoch_start"] *= 1e-9
                trigger["epoch_end"] *= 1e-9
            event["event"]["time"] *= 1e-9
        yield event

missed_found_by_on_ifos(far_threshold=1 / 86400 / 365.25, segments_name='afterhtgate')

get missed and found instruments by on ifos FIXME I am sure this is stupidly slow

Source code in sgnl/sgnlio.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def missed_found_by_on_ifos(
    self, far_threshold=1 / 86400 / 365.25, segments_name="afterhtgate"
):
    """
    get missed and found instruments by on ifos
    FIXME I am sure this is stupidly slow
    """

    _missed, _found = self.get_missed_found(
        selection_func=lambda r: r["event"]["combined_far"] <= far_threshold
    )
    _segments = self.segments(name=segments_name)

    missed = _MissedByOnIFOs(_segments, _missed, self.schema)
    found = _FoundByOnIFOs(_segments, _found, self.schema)

    return missed, found

segments(name='afterhtgate')

Retrieve segments with the specified name from the database and organize them into a dictionary by interferometer (ifo) combination.

Source code in sgnl/sgnlio.py 
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
def segments(self, name="afterhtgate"):
    """
    Retrieve segments with the specified name from the database and organize them
    into a dictionary by interferometer (ifo) combination.
    """
    segments = {}
    for row in self.db.cursor().execute(
        "SELECT * FROM segment WHERE name = ?", (name,)
    ):
        row = dict(row)
        segments.setdefault(row["ifo"], []).append(
            (row["start_time"], row["end_time"])
        )

    # Convert rows to segment list dictionary
    segments = segmentlistdict(
        {
            ifo: segmentlist(segment(*interval) for interval in intervals)
            for ifo, intervals in segments.items()
        }
    )

    # Generate all possible ifo combinations
    ifos = frozenset(segments.keys())
    combos = [
        frozenset(combo)
        for level in range(len(ifos), 0, -1)
        for combo in itertools.combinations(ifos, level)
    ]

    # Create the output dictionary
    out = {
        combo: segments.intersection(combo) - segments.union(ifos - combo)
        for combo in combos
    }

    return segmentlistdict(out)