framesrc_test_01.py

#!/usr/bin/env python
# Copyright (C) 2010  Kipp Cannon
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation; either version 2 of the License, or (at your
# option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
# Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.

#
# =============================================================================
#
#                                   Preamble
#
# =============================================================================
#


from gstlal import pipeparts
from gstlal.pipeparts import gst
from pylal.datatypes import LIGOTimeGPS
import test_common


#
# =============================================================================
#
#                                  Pipelines
#
# =============================================================================
#


#
# check consistency of buffer metadata generated by framesrc, and confirm
# that seeking and EOS work as expected
#


def framesrc_test_01a(pipeline, name):
    #
    # change these as needed to make test run on your machine
    #

    location = "/home/kipp/scratch_local/874100000-20000/cache/874000000-20000.cache"
    instrument = "H1"
    channel_name = "LSC-STRAIN"

    #
    # build pipeline
    #

    head = pipeparts.mkframecppchanneldemux(pipeline, pipeparts.mkcachesrc(pipeline, location = location, cache_src_regex = "%s.*" % instrument[0]))
    elem = pipeparts.mkqueue(pipeline, None, max_size_time = 8 * gst.SECOND)
    pipeparts.src_deferred_link(head, "%s:%s" % (instrument, channel_name), elem.get_pad("sink"))
    head = elem

    head = pipeparts.mkprogressreport(pipeline, head, "src")
    head = pipeparts.mkchecktimestamps(pipeline, head)
    pipeparts.mkfakesink(pipeline, head)

    #
    # done
    #

    return pipeline


#
# =============================================================================
#
#                                     Main
#
# =============================================================================
#


test_common.build_and_run(framesrc_test_01a, "framesrc_test_01a", segment = (LIGOTimeGPS(874000016), LIGOTimeGPS(874000600)))