Skip to content

sgnl.sgnlio

SgnlDB dataclass

Bases: StillSuit

Source code in sgnl/sgnlio.py 
 12
 13
 14
 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
 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
@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, template_duration=False, **kwargs):
        """
        A wrapper function of StillSuit.get_events() with additional
        functionalities
        """

        for event in super(SgnlDB, self).get_events(**kwargs):
            if template_duration:
                # Assume _filter_id is the same for all triggers in an event
                template_duration = dict(
                    self.db.cursor()
                    .execute(
                        "SELECT template_duration FROM filter WHERE _filter_id = ?",
                        (event["trigger"][0]["_filter_id"],),
                    )
                    .fetchone()
                )["template_duration"]
                for trigger in event["trigger"]:
                    trigger["template_duration"] = template_duration
            if nanosec_to_sec:
                for trigger in event["trigger"]:
                    trigger["time"] *= 1e-9
                    trigger["epoch_start"] *= 1e-9
                    trigger["epoch_end"] *= 1e-9
                    if "template_duration" in trigger:
                        trigger["template_duration"] *= 1e-9
                event["event"]["time"] *= 1e-9
            yield event

get_events(nanosec_to_sec=False, template_duration=False, **kwargs)

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

Source code in sgnl/sgnlio.py 
 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
def get_events(self, nanosec_to_sec=False, template_duration=False, **kwargs):
    """
    A wrapper function of StillSuit.get_events() with additional
    functionalities
    """

    for event in super(SgnlDB, self).get_events(**kwargs):
        if template_duration:
            # Assume _filter_id is the same for all triggers in an event
            template_duration = dict(
                self.db.cursor()
                .execute(
                    "SELECT template_duration FROM filter WHERE _filter_id = ?",
                    (event["trigger"][0]["_filter_id"],),
                )
                .fetchone()
            )["template_duration"]
            for trigger in event["trigger"]:
                trigger["template_duration"] = template_duration
        if nanosec_to_sec:
            for trigger in event["trigger"]:
                trigger["time"] *= 1e-9
                trigger["epoch_start"] *= 1e-9
                trigger["epoch_end"] *= 1e-9
                if "template_duration" in trigger:
                    trigger["template_duration"] *= 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 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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 
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
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)