def calc_inj_snrs(self, inj):
    """
    Estimate injected SNRs given injection time,
    waveform, intrinsic and extrinsic parameters.

    Parameters
    ----------
    inj (ligolw table row)
        SimInspiral table row object corresponding to
        a single injection.

    Returns
    ----------
    snr (dict)
        Dictionary, keyed by ifo, of estimated injected
        SNRs.
    """
    snr = dict.fromkeys(self.ifos, 0.0)

    injtime = inj.geocent_end_time
    f_min = inj.f_lower
    approximant = lalsimulation.GetApproximantFromString(str(inj.waveform))
    sample_rate = 16384.0
    f_max = self.f_max

    h_plus, h_cross = lalsimulation.SimInspiralTD(
        m1=inj.mass1 * lal.MSUN_SI,
        m2=inj.mass2 * lal.MSUN_SI,
        S1x=inj.spin1x,
        S1y=inj.spin1y,
        S1z=inj.spin1z,
        S2x=inj.spin2x,
        S2y=inj.spin2y,
        S2z=inj.spin2z,
        distance=inj.distance * 1e6 * lal.PC_SI,
        inclination=inj.inclination,
        phiRef=inj.coa_phase,
        longAscNodes=0.0,
        eccentricity=0.0,
        meanPerAno=0.0,
        deltaT=1.0 / sample_rate,
        f_min=f_min,
        f_ref=0.0,
        LALparams=None,
        approximant=approximant,
    )

    h_plus.epoch += injtime
    h_cross.epoch += injtime

    # Compute strain in each detector. If one detector wasn't on,
    # snr will be set to zero.
    for instrument in snr:
        if instrument not in self.psd.keys():
            continue
        h = lalsimulation.SimDetectorStrainREAL8TimeSeries(
            h_plus,
            h_cross,
            inj.longitude,
            inj.latitude,
            inj.polarization,
            lalsimulation.DetectorPrefixToLALDetector(instrument),
        )
        snr[instrument] = lalsimulation.MeasureSNR(
            h, self.psd[instrument], f_min, f_max
        )

    return snr

@staticmethod
def mc_eta_from_m1_m2(m1, m2):
    """
    Compute chirp mass and mass ratio from component masses.
    """
    mc = (m1 * m2) ** (3.0 / 5.0) / (m1 + m2) ** (1.0 / 5.0)
    eta = (m1 * m2) / (m1 + m2) ** 2.0

    return mc, eta

def produce_coinc_message(self, trigger, coincfar):
    """
    Send output messages of simulated coinc events to output
    `inj_events` topic.

    Parameters
    ----------
    trigger (ligowl table row)
        Row corresponding to a single injection in the SimInspiral
        table.
    coincfar (float)
        FAR value for simulated recovered event associated with
        this injection.
    """
    # build coinc xml doc, calculate p_astro, and produce message
    newxmldoc = self.produce_output_coinc(trigger, coincfar)
    if not newxmldoc:
        return

    coinctable = lsctables.CoincInspiralTable.get_table(newxmldoc)
    coincrow = coinctable[0]

    time = coincrow.end_time
    coincsnr = coincrow.snr
    mchirp = coincrow.mchirp
    p_astro = pastro_utils.p_astro(mchirp, coincsnr, self.p_x_c, self.p_c)

    # write coinc file to disk
    filename = f"fake_coinc-{int(time)}.xml"
    ligolw_utils.write_filename(
        newxmldoc,
        os.path.join(self.coinc_output, filename),
        verbose=self.verbose
    )

    coinc_msg = io.BytesIO()
    ligolw_utils.write_fileobj(newxmldoc, coinc_msg)

    # create json packet
    output = {
        "time": time,
        "time_ns": coincrow.end_time_ns,
        "snr": coincsnr,
        "far": coincrow.combined_far,
        "p_astro": json.dumps(p_astro),
        "coinc": coinc_msg.getvalue().decode(),
    }

    # send coinc message to events topic
    logging.info(f"network SNR: {output['snr']} | FAR: {output['far']}")

    topic = f"fake-data.{self.tag}.testsuite.inj_events"
    self.producer.write(topic, output)
    logging.info(f"Sent msg to: {topic}")

    newxmldoc.unlink()

    return

def produce_output_coinc(self, row, coincfar):
    """
    Construct a full ligolw file object to represent a simulated
    recovered event.

    Parameters
    ----------
    row (ligolw table row)
        A single SimInspiral row corresponding to the injection.
    coincfar (float)
        FAR value for simulated recovered event associated with
        this injection.

    Returns
    ----------
    newxmldoc (ligolw file object)
        ligo-lw coinc file object with all required tables.
    """
    # instantiate relevant lsctables objects
    newxmldoc = ligolw.Document()
    ligolw_elem = newxmldoc.appendChild(ligolw.LIGO_LW())
    new_process_table = ligolw_elem.appendChild(
        lsctables.New(lsctables.ProcessTable,
                      columns=utils.all_process_rows)
    )
    new_sngl_inspiral_table = ligolw_elem.appendChild(
        lsctables.New(lsctables.SnglInspiralTable,
                      columns=utils.all_sngl_rows)
    )
    new_coinc_inspiral_table = ligolw_elem.appendChild(
        lsctables.New(lsctables.CoincInspiralTable,
                      columns=utils.all_coinc_rows)
    )
    new_coinc_event_table = ligolw_elem.appendChild(
        lsctables.New(lsctables.CoincTable)
    )
    new_coinc_map_table = ligolw_elem.appendChild(
        lsctables.New(lsctables.CoincMapTable)
    )

    # simulate SNR time series using interpolated psd object
    # measurement_error is set as gaussian but one can switch
    # to no noise by measurement_error="zero-noise"
    bayestar_sim_list = bayestar_realize_coincs.simulate(
        seed=None,
        sim_inspiral=row,
        psds=self.psds_interp,
        responses=self.responses,
        locations=self.locations,
        measurement_error="gaussian-noise",
        f_low=20,
        f_high=2048,
    )

    # get mass parameters
    mass1 = max(numpy.random.normal(loc=row.mass1, scale=1.0), 1.1)
    mass2 = max(numpy.random.normal(loc=row.mass2, scale=1.0), mass1)
    mchirp, eta = self.mc_eta_from_m1_m2(mass1, mass2)

    snrs = defaultdict(lambda: 0)
    coincsnr = None

    # populate process table
    process_row_dict = {k: 0 for k in utils.all_process_rows}
    process_row_dict.update(
        {"process_id": 0, "program": "gstlal_inspiral", "comment": ""}
    )
    new_process_table.extend([
        lsctables.ProcessTable.RowType(**process_row_dict)
    ])

    # populate sngl table, coinc map table, and SNR timeseriess
    for event_id, (
        ifo, (horizon, abs_snr, arg_snr, toa, snr_series)
    ) in enumerate(zip(self.ifos, bayestar_sim_list)):
        sngl_row_dict = {k: 0 for k in utils.all_sngl_rows}

        sngl_row_dict.update(
            {
                "process_id": 0,
                "event_id": event_id,
                "end": toa,
                "mchirp": mchirp,
                "mass1": mass1,
                "mass2": mass2,
                "eta": eta,
                "ifo": ifo,
                "snr": abs_snr,
                "coa_phase": arg_snr,
            }
        )

        # add to the sngl inspiral table
        new_sngl_inspiral_table.extend(
            [lsctables.SnglInspiralTable.RowType(**sngl_row_dict)]
        )
        snrs[ifo] = abs_snr

        coinc_map_row_dict = {
            "coinc_event_id": 0,
            "event_id": event_id,
            "table_name": "sngl_inspiral",
        }

        # add to the coinc map table
        new_coinc_map_table.extend(
            [lsctables.CoincMapTable.RowType(**coinc_map_row_dict)]
        )

        # add SNR time series as array objects
        elem = lal.series.build_COMPLEX8TimeSeries(snr_series)
        elem.appendChild(Param.from_pyvalue("event_id", event_id))
        ligolw_elem.appendChild(elem)

    # calculate coinc SNR, only proceed if above 4
    coincsnr = numpy.linalg.norm([snr for snr in snrs.values() if snr > 4])
    if not coincsnr:
        logging.debug(f"Coinc SNR {coincsnr} too low to send a message.")
        return None

    # populate coinc inspiral table
    coinc_row_dict = {col: 0 for col in utils.all_coinc_rows}
    coincendtime = row.geocent_end_time
    coincendtimens = row.geocent_end_time_ns
    coinc_row_dict.update(
        {
            "coinc_event_id": 0,
            "snr": coincsnr,
            "mass": row.mass1 + row.mass2,
            "mchirp": row.mchirp,
            "end_time": coincendtime,
            "end_time_ns": coincendtimens,
            "combined_far": coincfar,
        }
    )
    new_coinc_inspiral_table.extend(
        [lsctables.CoincInspiralTable.RowType(**coinc_row_dict)]
    )

    # populate coinc event table
    coinc_event_row_dict = {col: 0 for col in utils.all_coinc_event_rows}
    coinc_event_row_dict.update(
        {
            "coinc_def_id": 0,
            "process_id": 0,
            "time_slide_id": 0,
            "instruments": "H1,L1,V1",
            "numevents": len(new_sngl_inspiral_table),
        }
    )
    new_coinc_event_table.extend(
        [lsctables.CoincTable.RowType(**coinc_event_row_dict)]
    )

    # add psd frequeny series
    lal.series.make_psd_xmldoc(self.psd, ligolw_elem)

    return newxmldoc

def shift_times(self, row, time_offset):
    """
    fix RA and GPS times according to the time offset

    Parameters
    ----------
    row (ligolw table row)
        SimInspiral table row object corresponding to
        a single injection.
    time_offset (int)
        Offset to shift injection times by.

    Returns
    ----------
    row (ligolw table row)
        SimInspiral table row with shifted times and
        corrected right ascension
    """
    end_time = row.geocent_end_time + row.geocent_end_time_ns * 10.0**-9.0
    gmst0 = GreenwichMeanSiderealTime(LIGOTimeGPS(end_time))
    gmst = GreenwichMeanSiderealTime(LIGOTimeGPS(end_time + time_offset))
    dgmst = gmst - gmst0
    row.longitude = row.longitude + dgmst

    row.geocent_end_time = int(row.geocent_end_time + time_offset)
    row.h_end_time = row.h_end_time + time_offset
    row.l_end_time = row.l_end_time + time_offset
    row.v_end_time = row.v_end_time + time_offset

    return row