sgnl.bin.ll_inspiral_event_uploader

an executable to aggregate and upload GraceDB events from sgnl-inspiral jobs

EventUploader

Bases: EventProcessor

manages handling of incoming events, selecting the best and uploading to GraceDB.

Source code in sgnl/bin/ll_inspiral_event_uploader.py

class EventUploader(events.EventProcessor):
    """
    manages handling of incoming events, selecting the best and uploading to GraceDB.
    """

    _name = "event_uploader"

    def __init__(
        self,
        input_topic,
        kafka_server,
        logger,
        scald_config,
        far_threshold: float = 3.84e-07,
        far_trials_factor: int = 1,
        gracedb_group: str = "Test",
        gracedb_pipeline: str = "SGNL",
        gracedb_search: str = "LowMass",
        gracedb_service_url: str = DEFAULT_GRACEDB_URL,
        gracedb_kafka_server: str | None = None,
        gracedb_kafka_ca_cert: str | None = None,
        gracedb_kafka_skip_hostname_verification: bool = False,
        gracedb_netrc_auth: bool = False,
        max_event_time: int = 7200,
        max_partitions: int = 10,
        num_jobs: int = 10000,
        processing_cadence: float = 0.1,
        request_timeout: float = 0.2,
        selection_criteria: str = "MAXSNR",
        tag: str = "test",
        upload_cadence_factor: float = 4,
        upload_cadence_type: str = "geometric",
    ):
        self.logger = logger
        self.logger.info("setting up event uploader...")

        self.is_injection_job = input_topic == "inj_events"
        topic_prefix = "" if not self.is_injection_job else "inj_"
        heartbeat_topic = f"sgnl.{tag}.{topic_prefix}event_uploader_heartbeat"
        self.favored_event_topic = f"sgnl.{tag}.{topic_prefix}favored_events"
        self.upload_topic = f"sgnl.{tag}.{topic_prefix}uploads"

        # set up output topics. Note that the uploads topic is special,
        # we use it for the SNR optimizer. We want to divide the work
        # among multiple consumers so we set the number of partitions to
        # 10. This is calculated based on the number of expected g-events
        # within a 5 minute window. FIXME: check this number
        output_topics = [self.favored_event_topic, self.upload_topic]
        self.max_partitions = max_partitions
        num_partitions = [1, self.max_partitions]

        events.EventProcessor.__init__(
            self,
            process_cadence=processing_cadence,
            request_timeout=request_timeout,
            num_messages=num_jobs,
            kafka_server=kafka_server,
            input_topic=f"sgnl.{tag}.{input_topic}",
            output_topic=output_topics,
            topic_partitions=num_partitions,
            tag=tag,
            send_heartbeats=True,
            heartbeat_cadence=60.0,
            heartbeat_topic=heartbeat_topic,
        )

        # initialize timing options
        self.max_event_time = max_event_time
        self.retries = 5
        self.retry_delay = 1

        # initialize gracedb client and, for v2 uploads, a Kafka producer.
        # Events are created via the GraceDB v2 API: createEvent() reserves a
        # graceid over HTTP and then publishes the coinc to the GraceDB Kafka
        # broker. With http_fallback=True the client falls back to the v2 HTTP
        # upload path if reservation or Kafka delivery fails.
        self.kafka_producer = None
        # v2-only createEvent arguments are passed unless the netrc
        # basic-auth path selected the v1 client
        self.client_api_v2 = not gracedb_netrc_auth
        if gracedb_service_url.startswith("file"):
            self.client = FakeGracedbClient(gracedb_service_url)
        else:
            # Modern ligo-gracedb's credential discovery covers X.509
            # and SciTokens but not the HTTP basic auth used by
            # local/dev GraceDB deployments (e.g. the sgn-sdk container
            # stack).  The legacy client supported basic auth through
            # the netrc file; --netrc-auth opts in to that behavior:
            # look the service host up with safe_netrc (which honors
            # the NETRC environment variable and enforces file
            # permissions) and attach the credentials.  Without the
            # flag this code path never runs and the client behaves
            # exactly as before.
            if gracedb_netrc_auth:
                from safe_netrc import netrc

                host = urlparse(gracedb_service_url).hostname
                try:
                    auth = netrc().authenticators(host)
                except OSError as err:
                    raise ValueError(
                        "--netrc-auth was given but the netrc file could "
                        f"not be read: {err}"
                    )
                if auth is None:
                    raise ValueError(
                        "--netrc-auth was given but the netrc file has "
                        f"no entry for host '{host}'"
                    )
                username, _, password = auth
                self.logger.info("using netrc basic auth credentials for GraceDB")
                # the GraceDB server only honors basic auth on the v1
                # API surface (its basic-auth backend is scoped to the
                # v1 app), so the netrc path must use the v1 client
                client = GraceDb(
                    gracedb_service_url, api_version="v1", force_noauth=True
                )
                creds = base64.b64encode(f"{username}:{password}".encode()).decode()
                client.headers["Authorization"] = f"Basic {creds}"
            else:
                client = GraceDb(gracedb_service_url, api_version="v2")
            self.client = client
            if gracedb_kafka_server:
                # producing events needs the kafka.produce scope;
                # create_kafka_producer() otherwise defaults to gracedb.read
                kafka_produce_scope = "kafka.produce"
                self.kafka_producer = client.create_kafka_producer(
                    bootstrap_servers=gracedb_kafka_server,
                    ca_cert_path=gracedb_kafka_ca_cert,
                    skip_hostname_verification=(
                        gracedb_kafka_skip_hostname_verification
                    ),
                    token_scope=kafka_produce_scope,
                )

        # gracedb settings
        self.gracedb_group = gracedb_group
        self.gracedb_pipeline = gracedb_pipeline
        self.gracedb_search = gracedb_search

        # upload cadence settings
        self.upload_cadence_type = upload_cadence_type
        self.upload_cadence_factor = upload_cadence_factor

        # initialize event store
        self.events: OrderedDict = OrderedDict()

        # favored event settings
        if selection_criteria == "MAXSNR":
            self.favored_function = self.select_maxsnr_candidate
        elif selection_criteria == "MINFAR":
            self.favored_function = self.select_minfar_candidate
        else:
            self.favored_function = self.construct_composite_candidate

        self.public_far_threshold = far_threshold / far_trials_factor

        # heartbeat settings
        self.last_inspiral_heartbeat = 0.0
        self.heartbeat_write = utils.gps_now()

        # upload topic settings
        # keep track of the partition that we send messages to
        # so that we can iterate across all partitions evenly
        # start with 0, iterate up to self.max_partitions - 1, and repeat
        self.partition_key = 0

        # set up aggregator sink
        with open(scald_config, "r") as f:
            agg_config = yaml.safe_load(f)
        self.agg_sink = influx.Aggregator(**agg_config["backends"]["default"])

        # register measurement schemas for aggregators
        self.agg_sink.load(path=scald_config)

    def ingest(self, message):
        """
        parse a message containing a candidate event
        """
        # process heartbeat messages from inspiral jobs
        if message.key() and "heartbeat" == message.key().decode("UTF-8"):
            heartbeat = json.loads(message.value())
            if heartbeat["time"] > self.last_inspiral_heartbeat:
                self.last_inspiral_heartbeat = heartbeat["time"]

        # process candidate event
        else:
            candidate = json.loads(message.value())
            candidate["time"] = LIGOTimeGPS(candidate["time"], candidate.pop("time_ns"))
            candidate.update(self.trigger_info(candidate))
            self.process_candidate(candidate)

    def process_candidate(self, candidate):
        """
        handles the processing of a candidate, creating
        a new event if necessary
        """
        key = self.event_window(candidate["time"])
        if key in self.events:
            self.logger.info("adding new candidate for event: [%.1f, %.1f]", *key)
            self.events[key]["candidates"].append(candidate)
            self.update_trigger_history(key, candidate)
        else:
            new_event = True
            for seg, event in self.events.items():
                if segment(candidate["time"], candidate["time"]) in seg:
                    self.logger.info(
                        "adding new candidate for time window: [%.1f, %.1f]", *seg
                    )
                    event["candidates"].append(candidate)
                    self.update_trigger_history(seg, candidate)
                    new_event = False

            # event not found, create a new event
            if new_event:
                self.logger.info("found new event: [%.1f, %.1f]", *key)
                self.events[key] = self.new_event()
                self.events[key]["candidates"].append(candidate)
                self.update_trigger_history(key, candidate)

    def update_trigger_history(self, key, candidate):
        """
        update trigger history dict for each candidate
        """
        self.events[key]["trigger_history"].update(candidate["trigger_info"])

    def trigger_info(self, candidate):
        """
        gather trigger information for each candidate,
        used for rtpe
        """
        # parse candidate for SVD bin, masses, LR, and SNR
        svdbin, type = self.parse_job_tag(candidate["job_tag"])

        coinc = self.load_xmlobj(candidate["coinc"])
        coinc_row = lsctables.CoincTable.get_table(coinc)[0]
        snglinspiral_row = lsctables.SnglInspiralTable.get_table(coinc)[0]

        return {
            "trigger_info": {
                svdbin: {
                    "snr": candidate["snr"],
                    "likelihood": coinc_row.likelihood,
                    "mass1": snglinspiral_row.mass1,
                    "mass2": snglinspiral_row.mass2,
                    "spin1z": snglinspiral_row.spin1z,
                    "spin2z": snglinspiral_row.spin2z,
                    "Gamma0": snglinspiral_row.Gamma0,
                }
            }
        }

    def parse_job_tag(self, tag):
        """
        get svd bin and job type from job tag
        """
        svdbin = tag.split("_")[0]
        name = "_".join(tag.split("_")[1:])
        return svdbin, name

    def load_xmlobj(self, xmlobj):
        """
        returns the coinc xml object from the kafka message
        """
        if isinstance(xmlobj, str):
            xmlobj = BytesIO(xmlobj.encode("utf-8"))
        return ligolw_utils.load_fileobj(xmlobj, contenthandler=LIGOLWContentHandler)

    def event_window(self, t):
        """
        returns the event window representing the event
        """
        dt = 0.2
        return segment(utils.floor_div(t - dt, 0.5), utils.floor_div(t + dt, 0.5) + 0.5)

    def new_event(self):
        """
        returns the structure that defines an event
        """
        return {
            "num_sent": 0,
            "time_sent": None,
            "favored": None,
            "gid": None,
            "candidates": deque(maxlen=self.num_messages),
            "trigger_history": {},
        }

    def handle(self):
        """
        handle events stored, selecting the best candidate.
        upload if a new favored event is found
        """
        for key, event in sorted(self.events.items(), reverse=True):
            if (
                (event["num_sent"] == 0 and len(event["candidates"]) > 0)
                or event["candidates"]
                and (
                    (utils.gps_now() >= self.next_event_upload(event))
                    or (
                        event["favored"]["far"] > self.public_far_threshold
                        and any(
                            [
                                candidate["far"] <= self.public_far_threshold
                                for candidate in event["candidates"]
                            ]
                        )
                    )
                )
            ):
                self.logger.info(
                    "handle: process_event num %d [%.1f, %.1f]",
                    len(event["candidates"]),
                    *key,
                )
                self.process_event(event, key)

        # clean out old events
        current_time = utils.gps_now()
        for key in list(self.events.keys()):
            if current_time - key[0] >= self.max_event_time:
                if self.events[key]["gid"]:
                    self.logger.info(
                        "sending final trigger history for event [%.1f, %.1f]", *key
                    )
                    self.upload_file(
                        "Trigger history file for RTPE",
                        "trigger_history.json",
                        "trigger_history",
                        json.dumps(self.events[key]["trigger_history"]),
                        self.events[key]["gid"],
                    )
                self.logger.info("removing stale event [%.1f, %.1f]", *key)
                self.events.pop(key)

        # has it been more than 15 minutes since last inspiral heartbeat
        # jobs should send heartbeats every 10 minutes
        now = utils.gps_now()
        if now - self.heartbeat_write > 15 * 60:
            state = 0 if now - self.last_inspiral_heartbeat > 15 * 60 else 1
            data = {
                "heartbeat": {
                    "heartbeat_tag": {"time": [int(now)], "fields": {"data": [state]}}
                }
            }

            self.logger.debug("Storing heartbeat state %d to influx...", state)
            self.agg_sink.store_columns("heartbeat", data["heartbeat"], aggregate=None)
            self.heartbeat_write = now

    def process_event(self, event, window):
        """
        handle a single event, selecting the best candidate.
        upload if a new favored event is found
        """
        updated, event = self.process_candidates(event)
        if event["num_sent"] == 0:
            assert updated
        if updated:
            self.logger.info(
                "uploading %s candidate with FAR = %.3E, "
                "SNR = %2.1f for event: [%.1f, %.1f]",
                self.to_ordinal(event["num_sent"] + 1),
                event["favored"]["far"],
                event["favored"]["snr"],
                window[0],
                window[1],
            )
            gid = self.upload_event(event)
            self.send_favored_event(event, window)
            if gid:
                event["num_sent"] += 1
                event["gid"] = gid
                self.send_uploaded(event, gid)

    def process_candidates(self, event):
        """
        process candidates and update the favored (maxsnr) event
        if needed

        returns event and whether the favored (maxsnr) event was updated
        """
        updated, this_favored = self.favored_function(
            event["candidates"], event["favored"]
        )

        if updated:
            event["favored"] = this_favored

        event["candidates"].clear()

        return updated, event

    def construct_composite_candidate(self, candidates, favored=None):
        """
        Construct composite event. Replace far and likelihood
        in maxsnr candidate with those in the minfar candidate.
        """

        # add previous favored event to list so we can get
        # the overall min FAR and max SNR instead of just
        # over the new candidates
        if favored:
            candidates.append(favored)

        maxsnr_candidate = max(candidates, key=self.rank_snr)
        maxsnr = maxsnr_candidate["snr"]

        minfar_candidate = min(candidates, key=self.rank_far)
        minfar = minfar_candidate["far"]

        # if neither the FAR nor SNR have improved compared to
        # the previous favored, no update. Otherwise, make
        # composite event and send an update
        if favored and maxsnr <= favored["snr"] and minfar >= favored["far"]:
            return False, favored
        else:
            # construct composite event
            if maxsnr_candidate["far"] != minfar:
                self.logger.info(
                    "construct new composite event with FAR: %.3E, " "SNR: %2.3f",
                    minfar,
                    maxsnr_candidate["snr"],
                )

                # replace far
                maxsnr_candidate["far"] = minfar

                # load coinc file
                maxsnr_coinc_row, maxsnr_coinc_file = self.get_coinc_row(
                    maxsnr_candidate
                )
                minfar_coinc_row, minfar_coinc_file = self.get_coinc_row(
                    minfar_candidate
                )

                # update likelihood and far
                maxsnr_coinc_row.likelihood = minfar_coinc_row.likelihood
                maxsnr_coinc_row.combined_far = minfar

                # save coinc file
                coinc_obj = BytesIO()
                ligolw_utils.write_fileobj(maxsnr_coinc_file, coinc_obj)
                maxsnr_candidate["coinc"] = coinc_obj.getvalue().decode("utf-8")

            if favored:
                assert (
                    maxsnr_candidate["far"] <= favored["far"]
                ), "composite event FAR should be smaller than previous favored FAR"
                assert (
                    maxsnr_candidate["snr"] >= favored["snr"]
                ), "composite event SNR should be larger than previous favored SNR"

            return True, maxsnr_candidate

    def select_maxsnr_candidate(self, candidates, favored=None):
        """
        select max snr candidate from candidates below
        public alert threshold:
        """
        # select the best candidate
        new_favored = max(candidates, key=self.rank_candidate)
        if not favored:
            return True, new_favored
        elif self.rank_candidate(new_favored) > self.rank_candidate(favored):
            return True, new_favored
        else:
            return False, favored

    def select_minfar_candidate(self, candidates, favored=None):
        """
        select the min far candidate out of the candidates
        """
        minfar_candidate = min(candidates, key=self.rank_far)
        if favored and minfar_candidate["far"] >= favored["far"]:
            return False, minfar_candidate
        else:
            return True, minfar_candidate

    def rank_candidate(self, candidate):
        """
        rank a candidate based on the following criterion:
        * FAR >  public threshold, choose lowest FAR
        * FAR <= public threshold, choose highest SNR
        """
        if candidate["far"] <= self.public_far_threshold:
            return True, candidate["snr"], 1.0 / candidate["far"]
        else:
            return False, 1.0 / candidate["far"], candidate["snr"]

    @staticmethod
    def rank_snr(candidate):
        return candidate["snr"]

    @staticmethod
    def rank_far(candidate):
        return candidate["far"]

    def send_favored_event(self, event, event_window):
        """
        send a favored event via Kafka
        """
        favored_event = {
            "event_window": list(event_window),
            "time": event["favored"]["time"].gpsSeconds,
            "time_ns": event["favored"]["time"].gpsNanoSeconds,
            "snr": event["favored"]["snr"],
            "far": event["favored"]["far"],
            "coinc": event["favored"]["coinc"],
        }
        self.producer.produce(
            topic=self.favored_event_topic, value=json.dumps(favored_event)
        )
        self.producer.poll(0)

    def send_uploaded(self, event, gid):
        """
        send an uploaded event via Kafka
        """
        uploaded = {
            "gid": gid,
            "time": event["favored"]["time"].gpsSeconds,
            "time_ns": event["favored"]["time"].gpsNanoSeconds,
            "snr": event["favored"]["snr"],
            "far": event["favored"]["far"],
            "coinc": event["favored"]["coinc"],
            "is_injection": self.is_injection_job,
            "snr_optimized": (
                True
                if "snr_optimized" in event["favored"]
                and event["favored"]["snr_optimized"]
                else False
            ),  # prevents re-triggering of the optimizer
        }

        # iterate the key so the next message goes to a different partition
        if self.partition_key >= self.max_partitions:
            self.partition_key = 0
        self.producer.produce(
            topic=self.upload_topic,
            value=json.dumps(uploaded),
            partition=self.partition_key,
        )
        self.partition_key += 1
        self.producer.poll(0)

    def upload_event(self, event):
        """
        upload a new event + auxiliary files
        """
        # upload event
        create_kwargs = dict(
            group=self.gracedb_group,
            pipeline=self.gracedb_pipeline,
            filename="coinc.xml",
            search=self.gracedb_search,
            offline=False,
        )
        if self.client_api_v2:
            # kafka= and http_fallback= only exist on the v2 client
            create_kwargs.update(kafka=self.kafka_producer, http_fallback=True)
        for attempt in range(1, self.retries + 1):
            try:
                resp = self.client.createEvent(
                    filecontents=event["favored"]["coinc"],
                    labels=(
                        "SNR_OPTIMIZED"
                        if "apply_snr_optimized_label" in event["favored"]
                        and event["favored"]["apply_snr_optimized_label"]
                        else None
                    ),
                    # don't apply the SNR_OPTIMIZED label for skymap optimizer events
                    **create_kwargs,
                )
            except HTTPError:
                self.logger.exception("upload_event:HTTPError")
            except Exception:
                self.logger.exception("upload_event:Exception")
            else:
                resp_json = resp.json()
                # The v2 Kafka path returns 201 from the graceid reservation;
                # an HTTP fallback to upload-reserved-event returns 200. Accept
                # both as success.  (v1 responses expose status_code.)
                status = getattr(resp, "status", None)
                if status is None:
                    status = getattr(resp, "status_code", None)
                if status in (httplib.OK, httplib.CREATED):
                    graceid = resp_json["graceid"]
                    self.logger.info("event assigned grace ID %s", graceid)
                    if not event["time_sent"]:
                        event["time_sent"] = utils.gps_now()
                    break
            self.logger.warning(
                "gracedb upload of %s " "failed on attempt %d/%d",
                "coinc.xml",
                attempt,
                self.retries,
            )
            time.sleep(numpy.random.lognormal(math.log(self.retry_delay), 0.5))
        else:
            self.logger.warning("gracedb upload of %s failed", "coinc.xml")
            return None

        self.upload_file(
            "Trigger history file for RTPE",
            "trigger_history.json",
            "trigger_history",
            json.dumps(event["trigger_history"]),
            graceid,
        )

        return graceid

    def upload_file(self, message, filename, tag, contents, graceid):
        """
        upload a file to gracedb
        """
        self.logger.info("posting %s to gracedb ID %s", filename, graceid)
        for attempt in range(1, self.retries + 1):
            try:
                resp = self.client.writeLog(
                    graceid,
                    message,
                    filename=filename,
                    filecontents=contents,
                    tagname=tag,
                )
            except HTTPError:
                self.logger.exception("upload_file:HTTPError")
            else:
                status = getattr(resp, "status", None)
                if status is None:
                    status = getattr(resp, "status_code", None)
                if status == httplib.CREATED:
                    break
            self.logger.warning(
                "gracedb upload of %s for ID %s " "failed on attempt %d/%d",
                filename,
                graceid,
                attempt,
                self.retries,
            )
            time.sleep(numpy.random.lognormal(math.log(self.retry_delay), 0.5))
        else:
            self.logger.warning(
                "gracedb upload of %s for ID %s failed", filename, graceid
            )

            return False

    def next_event_upload(self, event):
        """
        check whether enough time has elapsed to send an updated event
        """
        if self.upload_cadence_type == "geometric":
            return event["time_sent"] + numpy.power(
                self.upload_cadence_factor, event["num_sent"]
            )
        elif self.upload_cadence_type == "linear":
            return event["time_sent"] + self.upload_cadence_factor * event["num_sent"]

    def finish(self):
        """
        send remaining events before shutting down
        """
        for key, event in sorted(self.events.items(), reverse=True):
            if event["candidates"]:
                self.process_event(event, key)

        # flush and close the GraceDB Kafka producer
        if self.kafka_producer is not None:
            self.kafka_producer.close()

    @staticmethod
    def to_ordinal(n):
        """
        given an integer, returns the ordinal number
        representation.

        this black magic is taken from
        https://stackoverflow.com/a/20007730
        """
        return "%d%s" % (n, "tsnrhtdd"[(n / 10 % 10 != 1) * (n % 10 < 4) * n % 10 :: 4])

    def get_coinc_row(self, event):
        coinc_file = self.load_xmlobj(event["coinc"])
        return lsctables.CoincTable.get_table(coinc_file)[0], coinc_file

construct_composite_candidate(candidates, favored=None)

Construct composite event. Replace far and likelihood in maxsnr candidate with those in the minfar candidate.

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def construct_composite_candidate(self, candidates, favored=None):
    """
    Construct composite event. Replace far and likelihood
    in maxsnr candidate with those in the minfar candidate.
    """

    # add previous favored event to list so we can get
    # the overall min FAR and max SNR instead of just
    # over the new candidates
    if favored:
        candidates.append(favored)

    maxsnr_candidate = max(candidates, key=self.rank_snr)
    maxsnr = maxsnr_candidate["snr"]

    minfar_candidate = min(candidates, key=self.rank_far)
    minfar = minfar_candidate["far"]

    # if neither the FAR nor SNR have improved compared to
    # the previous favored, no update. Otherwise, make
    # composite event and send an update
    if favored and maxsnr <= favored["snr"] and minfar >= favored["far"]:
        return False, favored
    else:
        # construct composite event
        if maxsnr_candidate["far"] != minfar:
            self.logger.info(
                "construct new composite event with FAR: %.3E, " "SNR: %2.3f",
                minfar,
                maxsnr_candidate["snr"],
            )

            # replace far
            maxsnr_candidate["far"] = minfar

            # load coinc file
            maxsnr_coinc_row, maxsnr_coinc_file = self.get_coinc_row(
                maxsnr_candidate
            )
            minfar_coinc_row, minfar_coinc_file = self.get_coinc_row(
                minfar_candidate
            )

            # update likelihood and far
            maxsnr_coinc_row.likelihood = minfar_coinc_row.likelihood
            maxsnr_coinc_row.combined_far = minfar

            # save coinc file
            coinc_obj = BytesIO()
            ligolw_utils.write_fileobj(maxsnr_coinc_file, coinc_obj)
            maxsnr_candidate["coinc"] = coinc_obj.getvalue().decode("utf-8")

        if favored:
            assert (
                maxsnr_candidate["far"] <= favored["far"]
            ), "composite event FAR should be smaller than previous favored FAR"
            assert (
                maxsnr_candidate["snr"] >= favored["snr"]
            ), "composite event SNR should be larger than previous favored SNR"

        return True, maxsnr_candidate

event_window(t)

returns the event window representing the event

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def event_window(self, t):
    """
    returns the event window representing the event
    """
    dt = 0.2
    return segment(utils.floor_div(t - dt, 0.5), utils.floor_div(t + dt, 0.5) + 0.5)

finish()

send remaining events before shutting down

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def finish(self):
    """
    send remaining events before shutting down
    """
    for key, event in sorted(self.events.items(), reverse=True):
        if event["candidates"]:
            self.process_event(event, key)

    # flush and close the GraceDB Kafka producer
    if self.kafka_producer is not None:
        self.kafka_producer.close()

handle()

handle events stored, selecting the best candidate. upload if a new favored event is found

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def handle(self):
    """
    handle events stored, selecting the best candidate.
    upload if a new favored event is found
    """
    for key, event in sorted(self.events.items(), reverse=True):
        if (
            (event["num_sent"] == 0 and len(event["candidates"]) > 0)
            or event["candidates"]
            and (
                (utils.gps_now() >= self.next_event_upload(event))
                or (
                    event["favored"]["far"] > self.public_far_threshold
                    and any(
                        [
                            candidate["far"] <= self.public_far_threshold
                            for candidate in event["candidates"]
                        ]
                    )
                )
            )
        ):
            self.logger.info(
                "handle: process_event num %d [%.1f, %.1f]",
                len(event["candidates"]),
                *key,
            )
            self.process_event(event, key)

    # clean out old events
    current_time = utils.gps_now()
    for key in list(self.events.keys()):
        if current_time - key[0] >= self.max_event_time:
            if self.events[key]["gid"]:
                self.logger.info(
                    "sending final trigger history for event [%.1f, %.1f]", *key
                )
                self.upload_file(
                    "Trigger history file for RTPE",
                    "trigger_history.json",
                    "trigger_history",
                    json.dumps(self.events[key]["trigger_history"]),
                    self.events[key]["gid"],
                )
            self.logger.info("removing stale event [%.1f, %.1f]", *key)
            self.events.pop(key)

    # has it been more than 15 minutes since last inspiral heartbeat
    # jobs should send heartbeats every 10 minutes
    now = utils.gps_now()
    if now - self.heartbeat_write > 15 * 60:
        state = 0 if now - self.last_inspiral_heartbeat > 15 * 60 else 1
        data = {
            "heartbeat": {
                "heartbeat_tag": {"time": [int(now)], "fields": {"data": [state]}}
            }
        }

        self.logger.debug("Storing heartbeat state %d to influx...", state)
        self.agg_sink.store_columns("heartbeat", data["heartbeat"], aggregate=None)
        self.heartbeat_write = now

ingest(message)

parse a message containing a candidate event

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def ingest(self, message):
    """
    parse a message containing a candidate event
    """
    # process heartbeat messages from inspiral jobs
    if message.key() and "heartbeat" == message.key().decode("UTF-8"):
        heartbeat = json.loads(message.value())
        if heartbeat["time"] > self.last_inspiral_heartbeat:
            self.last_inspiral_heartbeat = heartbeat["time"]

    # process candidate event
    else:
        candidate = json.loads(message.value())
        candidate["time"] = LIGOTimeGPS(candidate["time"], candidate.pop("time_ns"))
        candidate.update(self.trigger_info(candidate))
        self.process_candidate(candidate)

load_xmlobj(xmlobj)

returns the coinc xml object from the kafka message

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def load_xmlobj(self, xmlobj):
    """
    returns the coinc xml object from the kafka message
    """
    if isinstance(xmlobj, str):
        xmlobj = BytesIO(xmlobj.encode("utf-8"))
    return ligolw_utils.load_fileobj(xmlobj, contenthandler=LIGOLWContentHandler)

new_event()

returns the structure that defines an event

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def new_event(self):
    """
    returns the structure that defines an event
    """
    return {
        "num_sent": 0,
        "time_sent": None,
        "favored": None,
        "gid": None,
        "candidates": deque(maxlen=self.num_messages),
        "trigger_history": {},
    }

next_event_upload(event)

check whether enough time has elapsed to send an updated event

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def next_event_upload(self, event):
    """
    check whether enough time has elapsed to send an updated event
    """
    if self.upload_cadence_type == "geometric":
        return event["time_sent"] + numpy.power(
            self.upload_cadence_factor, event["num_sent"]
        )
    elif self.upload_cadence_type == "linear":
        return event["time_sent"] + self.upload_cadence_factor * event["num_sent"]

parse_job_tag(tag)

get svd bin and job type from job tag

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def parse_job_tag(self, tag):
    """
    get svd bin and job type from job tag
    """
    svdbin = tag.split("_")[0]
    name = "_".join(tag.split("_")[1:])
    return svdbin, name

process_candidate(candidate)

handles the processing of a candidate, creating a new event if necessary

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def process_candidate(self, candidate):
    """
    handles the processing of a candidate, creating
    a new event if necessary
    """
    key = self.event_window(candidate["time"])
    if key in self.events:
        self.logger.info("adding new candidate for event: [%.1f, %.1f]", *key)
        self.events[key]["candidates"].append(candidate)
        self.update_trigger_history(key, candidate)
    else:
        new_event = True
        for seg, event in self.events.items():
            if segment(candidate["time"], candidate["time"]) in seg:
                self.logger.info(
                    "adding new candidate for time window: [%.1f, %.1f]", *seg
                )
                event["candidates"].append(candidate)
                self.update_trigger_history(seg, candidate)
                new_event = False

        # event not found, create a new event
        if new_event:
            self.logger.info("found new event: [%.1f, %.1f]", *key)
            self.events[key] = self.new_event()
            self.events[key]["candidates"].append(candidate)
            self.update_trigger_history(key, candidate)

process_candidates(event)

process candidates and update the favored (maxsnr) event if needed

returns event and whether the favored (maxsnr) event was updated

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def process_candidates(self, event):
    """
    process candidates and update the favored (maxsnr) event
    if needed

    returns event and whether the favored (maxsnr) event was updated
    """
    updated, this_favored = self.favored_function(
        event["candidates"], event["favored"]
    )

    if updated:
        event["favored"] = this_favored

    event["candidates"].clear()

    return updated, event

process_event(event, window)

handle a single event, selecting the best candidate. upload if a new favored event is found

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def process_event(self, event, window):
    """
    handle a single event, selecting the best candidate.
    upload if a new favored event is found
    """
    updated, event = self.process_candidates(event)
    if event["num_sent"] == 0:
        assert updated
    if updated:
        self.logger.info(
            "uploading %s candidate with FAR = %.3E, "
            "SNR = %2.1f for event: [%.1f, %.1f]",
            self.to_ordinal(event["num_sent"] + 1),
            event["favored"]["far"],
            event["favored"]["snr"],
            window[0],
            window[1],
        )
        gid = self.upload_event(event)
        self.send_favored_event(event, window)
        if gid:
            event["num_sent"] += 1
            event["gid"] = gid
            self.send_uploaded(event, gid)

rank_candidate(candidate)

rank a candidate based on the following criterion: * FAR > public threshold, choose lowest FAR * FAR <= public threshold, choose highest SNR

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def rank_candidate(self, candidate):
    """
    rank a candidate based on the following criterion:
    * FAR >  public threshold, choose lowest FAR
    * FAR <= public threshold, choose highest SNR
    """
    if candidate["far"] <= self.public_far_threshold:
        return True, candidate["snr"], 1.0 / candidate["far"]
    else:
        return False, 1.0 / candidate["far"], candidate["snr"]

select_maxsnr_candidate(candidates, favored=None)

select max snr candidate from candidates below public alert threshold:

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def select_maxsnr_candidate(self, candidates, favored=None):
    """
    select max snr candidate from candidates below
    public alert threshold:
    """
    # select the best candidate
    new_favored = max(candidates, key=self.rank_candidate)
    if not favored:
        return True, new_favored
    elif self.rank_candidate(new_favored) > self.rank_candidate(favored):
        return True, new_favored
    else:
        return False, favored

select_minfar_candidate(candidates, favored=None)

select the min far candidate out of the candidates

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def select_minfar_candidate(self, candidates, favored=None):
    """
    select the min far candidate out of the candidates
    """
    minfar_candidate = min(candidates, key=self.rank_far)
    if favored and minfar_candidate["far"] >= favored["far"]:
        return False, minfar_candidate
    else:
        return True, minfar_candidate

send_favored_event(event, event_window)

send a favored event via Kafka

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def send_favored_event(self, event, event_window):
    """
    send a favored event via Kafka
    """
    favored_event = {
        "event_window": list(event_window),
        "time": event["favored"]["time"].gpsSeconds,
        "time_ns": event["favored"]["time"].gpsNanoSeconds,
        "snr": event["favored"]["snr"],
        "far": event["favored"]["far"],
        "coinc": event["favored"]["coinc"],
    }
    self.producer.produce(
        topic=self.favored_event_topic, value=json.dumps(favored_event)
    )
    self.producer.poll(0)

send_uploaded(event, gid)

send an uploaded event via Kafka

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def send_uploaded(self, event, gid):
    """
    send an uploaded event via Kafka
    """
    uploaded = {
        "gid": gid,
        "time": event["favored"]["time"].gpsSeconds,
        "time_ns": event["favored"]["time"].gpsNanoSeconds,
        "snr": event["favored"]["snr"],
        "far": event["favored"]["far"],
        "coinc": event["favored"]["coinc"],
        "is_injection": self.is_injection_job,
        "snr_optimized": (
            True
            if "snr_optimized" in event["favored"]
            and event["favored"]["snr_optimized"]
            else False
        ),  # prevents re-triggering of the optimizer
    }

    # iterate the key so the next message goes to a different partition
    if self.partition_key >= self.max_partitions:
        self.partition_key = 0
    self.producer.produce(
        topic=self.upload_topic,
        value=json.dumps(uploaded),
        partition=self.partition_key,
    )
    self.partition_key += 1
    self.producer.poll(0)

to_ordinal(n) staticmethod

given an integer, returns the ordinal number representation.

this black magic is taken from https://stackoverflow.com/a/20007730

Source code in sgnl/bin/ll_inspiral_event_uploader.py

@staticmethod
def to_ordinal(n):
    """
    given an integer, returns the ordinal number
    representation.

    this black magic is taken from
    https://stackoverflow.com/a/20007730
    """
    return "%d%s" % (n, "tsnrhtdd"[(n / 10 % 10 != 1) * (n % 10 < 4) * n % 10 :: 4])

trigger_info(candidate)

gather trigger information for each candidate, used for rtpe

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def trigger_info(self, candidate):
    """
    gather trigger information for each candidate,
    used for rtpe
    """
    # parse candidate for SVD bin, masses, LR, and SNR
    svdbin, type = self.parse_job_tag(candidate["job_tag"])

    coinc = self.load_xmlobj(candidate["coinc"])
    coinc_row = lsctables.CoincTable.get_table(coinc)[0]
    snglinspiral_row = lsctables.SnglInspiralTable.get_table(coinc)[0]

    return {
        "trigger_info": {
            svdbin: {
                "snr": candidate["snr"],
                "likelihood": coinc_row.likelihood,
                "mass1": snglinspiral_row.mass1,
                "mass2": snglinspiral_row.mass2,
                "spin1z": snglinspiral_row.spin1z,
                "spin2z": snglinspiral_row.spin2z,
                "Gamma0": snglinspiral_row.Gamma0,
            }
        }
    }

update_trigger_history(key, candidate)

update trigger history dict for each candidate

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def update_trigger_history(self, key, candidate):
    """
    update trigger history dict for each candidate
    """
    self.events[key]["trigger_history"].update(candidate["trigger_info"])

upload_event(event)

upload a new event + auxiliary files

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def upload_event(self, event):
    """
    upload a new event + auxiliary files
    """
    # upload event
    create_kwargs = dict(
        group=self.gracedb_group,
        pipeline=self.gracedb_pipeline,
        filename="coinc.xml",
        search=self.gracedb_search,
        offline=False,
    )
    if self.client_api_v2:
        # kafka= and http_fallback= only exist on the v2 client
        create_kwargs.update(kafka=self.kafka_producer, http_fallback=True)
    for attempt in range(1, self.retries + 1):
        try:
            resp = self.client.createEvent(
                filecontents=event["favored"]["coinc"],
                labels=(
                    "SNR_OPTIMIZED"
                    if "apply_snr_optimized_label" in event["favored"]
                    and event["favored"]["apply_snr_optimized_label"]
                    else None
                ),
                # don't apply the SNR_OPTIMIZED label for skymap optimizer events
                **create_kwargs,
            )
        except HTTPError:
            self.logger.exception("upload_event:HTTPError")
        except Exception:
            self.logger.exception("upload_event:Exception")
        else:
            resp_json = resp.json()
            # The v2 Kafka path returns 201 from the graceid reservation;
            # an HTTP fallback to upload-reserved-event returns 200. Accept
            # both as success.  (v1 responses expose status_code.)
            status = getattr(resp, "status", None)
            if status is None:
                status = getattr(resp, "status_code", None)
            if status in (httplib.OK, httplib.CREATED):
                graceid = resp_json["graceid"]
                self.logger.info("event assigned grace ID %s", graceid)
                if not event["time_sent"]:
                    event["time_sent"] = utils.gps_now()
                break
        self.logger.warning(
            "gracedb upload of %s " "failed on attempt %d/%d",
            "coinc.xml",
            attempt,
            self.retries,
        )
        time.sleep(numpy.random.lognormal(math.log(self.retry_delay), 0.5))
    else:
        self.logger.warning("gracedb upload of %s failed", "coinc.xml")
        return None

    self.upload_file(
        "Trigger history file for RTPE",
        "trigger_history.json",
        "trigger_history",
        json.dumps(event["trigger_history"]),
        graceid,
    )

    return graceid

upload_file(message, filename, tag, contents, graceid)

upload a file to gracedb

Source code in sgnl/bin/ll_inspiral_event_uploader.py

def upload_file(self, message, filename, tag, contents, graceid):
    """
    upload a file to gracedb
    """
    self.logger.info("posting %s to gracedb ID %s", filename, graceid)
    for attempt in range(1, self.retries + 1):
        try:
            resp = self.client.writeLog(
                graceid,
                message,
                filename=filename,
                filecontents=contents,
                tagname=tag,
            )
        except HTTPError:
            self.logger.exception("upload_file:HTTPError")
        else:
            status = getattr(resp, "status", None)
            if status is None:
                status = getattr(resp, "status_code", None)
            if status == httplib.CREATED:
                break
        self.logger.warning(
            "gracedb upload of %s for ID %s " "failed on attempt %d/%d",
            filename,
            graceid,
            attempt,
            self.retries,
        )
        time.sleep(numpy.random.lognormal(math.log(self.retry_delay), 0.5))
    else:
        self.logger.warning(
            "gracedb upload of %s for ID %s failed", filename, graceid
        )

        return False