Source code for idq.io.reporters.ligolw

import sys

from ligo.lw import ligolw
from ligo.lw import lsctables
from ligo.lw import utils as ligolw_utils
from ligo.lw.utils import process as ligolw_process
from ligo.lw.utils import segments as ligolw_segments

from ... import hookimpl
from . import DiskReporter


class LIGOLWContentHandler(ligolw.LIGOLWContentHandler):
    pass


lsctables.use_in(LIGOLWContentHandler)


class LIGOLWSegmentReporter(DiskReporter):
    """
    a variant of DiskReporter that expects to only deal with ligolw segments in
    XML form.
    """

    _suffix = "xml"

    def _write(self, path, segs, **kwargs):
        xmldoc = ligolw.Document()
        xmldoc.appendChild(ligolw.LIGO_LW())

        # create process params
        params = {"rootdir": self.rootdir, "start": self.start, "end": self.end}
        process = ligolw_process.register_to_xmldoc(xmldoc, sys.argv[0], params)

        # store segments into xmldoc
        with ligolw_segments.LigolwSegments(xmldoc, process) as xmlsegments:
            xmlsegments.insert_from_segmentlistdict({"idq": segs}, "segments")
            xmlsegments.finalize(process)

        # save to disk
        ligolw_utils.write_filename(xmldoc, path)

    @classmethod
    def read(cls, path):
        xmlsegments = ligolw_utils.load_filename(path, LIGOLWContentHandler)
        segs = ligolw_segments.segmenttable_get_by_name(xmlsegments, path).coalesce()
        return segs["idq"]


@hookimpl
def get_reporters():
    return {"segment:ligolw": LIGOLWSegmentReporter}