gstlal_insertgap.c

/*
 * Copyright (C) 2016 Aaron Viets <aaron.viets@ligo.org>
 *
 * 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.
 */


/**
 * SECTION:gstlal_insertgap
 * @short_description: This element replaces undesired values, specified
 * by the property #bad_data_intervals, with gaps.
 *
 * The primary purpose of this element is to reject certain values
 * considered by the user to be "bad data." This can be done by flagging
 * that data as a gap, replacing it with a specified value, or both. The
 * criteria for bad data is specified by the array property
 * #bad_data_intervals. Array indices 0, 2, 4, etc.,
 * represent maxima, and array indices 1, 3, 5, etc., represent the 
 * corresponding minima. For example, if 
 * #bad_data_intervals = [0, 1, 2, 3],
 * then any values that fall in one of the closed intervals [-inf, 0],
 * [1, 2], or [3, inf] will be rejected and gapped and/or replaced
 * as specified by the user. To reject a single value, say zero,
 * #bad_data_intervals should be [-max_double, 0, 0, max_double].
 * If the data stream is complex, the real and and imaginary parts of
 * each input is tested, and if either is bad, the value is rejected.
 * The #bad_data_intervals and #replace_value properties are applied to
 * both real and imaginary parts.
 * This element also has the ability to fill in discontinuities if the
 * property #fill-discont is set to true. Presentation timestamps and
 * buffer offsets are adjusted as needed.
 */


/*
 * ============================================================================
 *
 *				  Preamble
 *
 * ============================================================================
 */


/*
 * stuff from C
 */


#include <math.h>
#include <string.h>
#include <complex.h>


/*
 * stuff from gobject/gstreamer
 */


#include <glib.h>
#include <gst/gst.h>
#include <gst/audio/audio.h>


/*
 * our own stuff
 */


#include <gstlal/gstlal.h>
#include <gstlal/gstlal_debug.h>
#include <gstlal/gstlal_audio_info.h>
#include <gstlal_insertgap.h>


/*
 * ============================================================================
 *
 *			   GStreamer Boiler Plate
 *
 * ============================================================================
 */


static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE(
	GST_BASE_TRANSFORM_SINK_NAME,
	GST_PAD_SINK,
	GST_PAD_ALWAYS,
	GST_STATIC_CAPS(
		"audio/x-raw, " \
		"rate = (int) [1, MAX], " \
		"channels = (int) [1, MAX], " \
		"format = (string) { "GST_AUDIO_NE(U32)", " GST_AUDIO_NE(F32) ", " GST_AUDIO_NE(F64) ", " GST_AUDIO_NE(Z64) ", " GST_AUDIO_NE(Z128) " }, " \
		"layout = (string) interleaved, " \
		"channel-mask = (bitmask) 0"
	)
);


static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE(
	GST_BASE_TRANSFORM_SRC_NAME,
	GST_PAD_SRC,
	GST_PAD_ALWAYS,
	GST_STATIC_CAPS(
		"audio/x-raw, " \
		"rate = (int) [1, MAX], " \
		"channels = (int) [1, MAX], " \
		"format = (string) { "GST_AUDIO_NE(U32)", " GST_AUDIO_NE(F32) ", " GST_AUDIO_NE(F64) ", " GST_AUDIO_NE(Z64) ", " GST_AUDIO_NE(Z128) " }, " \
		"layout = (string) interleaved, " \
		"channel-mask = (bitmask) 0"
	)
);


#define GST_CAT_DEFAULT gstlal_insertgap_debug
GST_DEBUG_CATEGORY_STATIC(GST_CAT_DEFAULT);


G_DEFINE_TYPE_WITH_CODE(
	GSTLALInsertGap,
	gstlal_insertgap,
	GST_TYPE_ELEMENT,
	GST_DEBUG_CATEGORY_INIT(GST_CAT_DEFAULT, "lal_insertgap", 0, "lal_insertgap element")
);


/*
 * ============================================================================
 *
 *			       Utilities
 *
 * ============================================================================
 */


#define DEFINE_CHECK_DATA(DTYPE) \
static gboolean check_data_ ## DTYPE(DTYPE *data, double *bad_data_intervals, int array_length, int num_checks, gboolean remove_nan, gboolean remove_inf) { \
	int i, j; \
	gboolean result = TRUE; \
	for(i = 0; i < num_checks; i++) { \
		if(bad_data_intervals) { \
			for(j = 0; j < array_length; j += 2) \
				result &= data[i] > bad_data_intervals[j] && data[i] < bad_data_intervals[j + 1]; \
		} \
		if(remove_nan) \
			result &= !isnan(data[i]); \
		if(remove_inf) \
			result &= !isinf(data[i]); \
	} \
	return result; \
}


DEFINE_CHECK_DATA(float);
DEFINE_CHECK_DATA(double);
DEFINE_CHECK_DATA(guint32);


#define DEFINE_PROCESS_INBUF(DTYPE,COMPLEX) \
static GstFlowReturn process_inbuf_ ## DTYPE ## COMPLEX(const DTYPE COMPLEX *indata, DTYPE COMPLEX *outdata, GSTLALInsertGap *element, gboolean sinkbuf_gap, gboolean sinkbuf_discont, guint64 sinkbuf_offset, guint64 sinkbuf_offset_end, GstClockTime sinkbuf_dur, GstClockTime sinkbuf_pts, gboolean complex_data) \
{ \
	GstFlowReturn result = GST_FLOW_OK; \
	guint64 blocks, max_block_length, missing_samples; \
	missing_samples = 0; \
 \
	/*
	 * First, deal with discontinuity if necessary
	 */ \
	if(element->fill_discont && (element->last_sinkbuf_offset_end != 0) && (sinkbuf_pts != element->last_sinkbuf_ets)) { \
 \
		/* Track discont length and number of zero-length buffers */ \
		element->discont_time += (sinkbuf_pts - element->last_sinkbuf_ets); \
		element->empty_bufs += (sinkbuf_dur ? 0 : 1); \
 \
		/* Find number of missing samples and max block length in samples */ \
		missing_samples = gst_util_uint64_scale_int_round(sinkbuf_pts - element->last_sinkbuf_ets, element->rate, 1000000000); \
		blocks = (sinkbuf_pts - element->last_sinkbuf_ets + element->block_duration - 1) / element->block_duration; /* ceil */ \
		g_assert_cmpuint(blocks, >, 0); /* make sure that the discont is not zero length */ \
		max_block_length = (missing_samples + blocks - 1) / blocks; /* ceil */ \
		g_assert_cmpuint(max_block_length, >, 0); \
 \
		/* Message for debugging */ \
		if(sinkbuf_dur && sinkbuf_offset != sinkbuf_offset_end) \
			GST_WARNING_OBJECT(element, "filling discontinuity lasting %f seconds (%lu samples) including %lu zero-length buffers and starting at %f seconds (offset %lu)", (((double) element->discont_time) / 1000000000.0), gst_util_uint64_scale_int_round(element->discont_time, element->rate, 1000000000), element->empty_bufs, (double) sinkbuf_pts / 1000000000.0 - (double) element->discont_time / 1000000000.0, sinkbuf_offset); \
 \
		guint standard_blocks = (guint) (missing_samples / max_block_length); \
		guint64 last_block_length = missing_samples % max_block_length; \
		DTYPE COMPLEX sample_value; \
		if(complex_data) \
			sample_value = (element->replace_value < G_MAXDOUBLE) ? ((DTYPE) (element->replace_value)) * (1 + I) : 0; \
		else \
			sample_value = (element->replace_value < G_MAXDOUBLE) ? ((DTYPE) (element->replace_value)) : 0; \
 \
		/* first make and push any buffers of size max_buffer_size */ \
		if(standard_blocks != 0) { \
			guint buffer_num; \
			for(buffer_num = 0; buffer_num < standard_blocks; buffer_num++) { \
				GstBuffer *discont_buf; \
				DTYPE COMPLEX *discont_buf_data; \
				discont_buf_data = g_malloc(max_block_length * element->channels * sizeof(DTYPE COMPLEX)); \
				guint sample_num; \
				for(sample_num = 0; sample_num < max_block_length * element->channels; sample_num++) { \
					*discont_buf_data = sample_value; \
					discont_buf_data++; \
				} \
				discont_buf = gst_buffer_new_wrapped((discont_buf_data - max_block_length * element->channels), max_block_length * element->channels * sizeof(DTYPE COMPLEX)); \
				if(G_UNLIKELY(!discont_buf)) { \
					GST_ERROR_OBJECT(element, "failure creating sub-buffer"); \
					result = GST_FLOW_ERROR; \
					goto done; \
				} \
 \
				/* set flags, caps, offset, and timestamps. */ \
				GST_BUFFER_OFFSET(discont_buf) = element->last_sinkbuf_offset_end + element->discont_offset + buffer_num * max_block_length; \
				GST_BUFFER_OFFSET_END(discont_buf) = GST_BUFFER_OFFSET(discont_buf) + max_block_length; \
				GST_BUFFER_PTS(discont_buf) = element->last_sinkbuf_ets + gst_util_uint64_scale_round(sinkbuf_pts - element->last_sinkbuf_ets, (guint64) buffer_num * max_block_length, missing_samples); \
				GST_BUFFER_DURATION(discont_buf) = element->last_sinkbuf_ets + gst_util_uint64_scale_round(sinkbuf_pts - element->last_sinkbuf_ets, ((guint64) buffer_num + 1) * max_block_length, missing_samples) - GST_BUFFER_PTS(discont_buf); \
				GST_BUFFER_FLAG_UNSET(discont_buf, GST_BUFFER_FLAG_DISCONT); \
				if(element->insert_gap) \
					GST_BUFFER_FLAG_SET(discont_buf, GST_BUFFER_FLAG_GAP); \
				else \
					GST_BUFFER_FLAG_UNSET(discont_buf, GST_BUFFER_FLAG_GAP); \
 \
				/* push buffer downstream */ \
				GST_DEBUG_OBJECT(element, "pushing sub-buffer %" GST_BUFFER_BOUNDARIES_FORMAT, GST_BUFFER_BOUNDARIES_ARGS(discont_buf)); \
				result = gst_pad_push(element->srcpad, discont_buf); \
				if(G_UNLIKELY(result != GST_FLOW_OK)) { \
					GST_WARNING_OBJECT(element, "push failed: %s", gst_flow_get_name(result)); \
					goto done; \
				} \
			} \
		} \
 \
		/* then make and push the remainder buffer */ \
		if(last_block_length != 0) { \
			GstBuffer *last_discont_buf; \
			DTYPE COMPLEX *last_discont_buf_data; \
			last_discont_buf_data = g_malloc(last_block_length * element->channels * sizeof(DTYPE COMPLEX)); \
			guint sample_num; \
			for(sample_num = 0; sample_num < last_block_length * element->channels; sample_num++) { \
				*last_discont_buf_data = sample_value; \
				last_discont_buf_data++; \
			} \
			last_discont_buf = gst_buffer_new_wrapped((last_discont_buf_data - last_block_length * element->channels), last_block_length * element->channels * sizeof(DTYPE COMPLEX)); \
			if(G_UNLIKELY(!last_discont_buf)) { \
				GST_ERROR_OBJECT(element, "failure creating sub-buffer"); \
				result = GST_FLOW_ERROR; \
				goto done; \
			} \
 \
			/* set flags, caps, offset, and timestamps. */ \
			GST_BUFFER_OFFSET(last_discont_buf) = element->last_sinkbuf_offset_end + element->discont_offset + missing_samples - last_block_length; \
			GST_BUFFER_OFFSET_END(last_discont_buf) = GST_BUFFER_OFFSET(last_discont_buf) + last_block_length; \
			GST_BUFFER_PTS(last_discont_buf) = element->last_sinkbuf_ets + gst_util_uint64_scale_round(sinkbuf_pts - element->last_sinkbuf_ets, missing_samples - last_block_length, missing_samples); \
			GST_BUFFER_DURATION(last_discont_buf) = sinkbuf_pts - GST_BUFFER_PTS(last_discont_buf); \
			GST_BUFFER_FLAG_UNSET(last_discont_buf, GST_BUFFER_FLAG_DISCONT); \
			if(element->insert_gap) \
				GST_BUFFER_FLAG_SET(last_discont_buf, GST_BUFFER_FLAG_GAP); \
			else \
				GST_BUFFER_FLAG_UNSET(last_discont_buf, GST_BUFFER_FLAG_GAP); \
 \
			/* push buffer downstream */ \
			GST_DEBUG_OBJECT(element, "pushing sub-buffer %" GST_BUFFER_BOUNDARIES_FORMAT, GST_BUFFER_BOUNDARIES_ARGS(last_discont_buf)); \
			result = gst_pad_push(element->srcpad, last_discont_buf); \
			if(G_UNLIKELY(result != GST_FLOW_OK)) { \
				GST_WARNING_OBJECT(element, "push failed: %s", gst_flow_get_name(result)); \
				goto done; \
			} \
		} \
		element->discont_offset += missing_samples; \
	} \
	if(!sinkbuf_dur) \
		goto done; \
 \
	element->discont_time = 0; \
	element->empty_bufs = 0; \
 \
	/*
	 * Now, use data on input buffer to make next output buffer(s)
	 */ \
	gboolean data_is_bad, srcbuf_gap, srcbuf_gap_next; \
	guint64 offset, current_srcbuf_length; \
	current_srcbuf_length = 0; \
 \
	/* compute length of incoming buffer and maximum block length in samples */ \
	blocks = (sinkbuf_dur + element->block_duration - 1) / element->block_duration; /* ceil */ \
	g_assert_cmpuint(blocks, >, 0); /* make sure that the sinkbuf is not zero length */ \
	guint64 length = sinkbuf_offset_end - sinkbuf_offset; \
	max_block_length = (length + blocks - 1) / blocks; /* ceil */ \
	g_assert_cmpuint(max_block_length, >, 0); \
 \
	/* Check first sample */ \
	data_is_bad = !check_data_ ## DTYPE((DTYPE *) indata, element->bad_data_intervals, element->array_length, (1 + (int) complex_data) * element->channels, element->remove_nan, element->remove_inf); \
	srcbuf_gap = (sinkbuf_gap && (!(element->remove_gap))) || ((element->insert_gap) && data_is_bad); \
	for(offset = 0; offset < length; offset++) { \
		data_is_bad = !check_data_ ## DTYPE((DTYPE *) indata, element->bad_data_intervals, element->array_length, (1 + (int) complex_data) * element->channels, element->remove_nan, element->remove_inf); \
		srcbuf_gap_next = (sinkbuf_gap && (!(element->remove_gap))) || ((element->insert_gap) && data_is_bad); \
		if(complex_data) \
			*outdata = (((element->replace_value) < G_MAXDOUBLE) && data_is_bad) ? ((DTYPE) (element->replace_value)) * (1 + I) : *indata; \
		else \
			*outdata = (((element->replace_value) < G_MAXDOUBLE) && data_is_bad) ? (DTYPE) (element->replace_value) : *indata; \
		current_srcbuf_length++; \
		indata += element->channels; \
		outdata += element->channels; \
 \
		/* 
		 * We need to push an output buffer if:
		 * 1) The number of samples to be output equals the maximum output buffer size
		 * 2) We have reached the end of the input buffer 
		 * 3) The output data changes from non-gap to gap or vice-versa
		 */ \
		if((current_srcbuf_length >= max_block_length) || (offset >= length - 1) || (srcbuf_gap_next != srcbuf_gap)) { \
			if((current_srcbuf_length > max_block_length) || (offset > length - 1)) \
				g_assert_not_reached(); \
			if(srcbuf_gap_next != srcbuf_gap) { \
				/*
				 * In this case, we don't want the most recent sample since its
				 * gap state is different. Put it on the next output buffer.
				 */ \
				offset--; \
				current_srcbuf_length--; \
				indata -= element->channels; \
				outdata -= element->channels; \
			} \
			GstBuffer *srcbuf; \
			DTYPE COMPLEX *srcbuf_data; \
			srcbuf_data = g_malloc(current_srcbuf_length * element->channels * sizeof(DTYPE COMPLEX)); \
			memcpy(srcbuf_data, (outdata - current_srcbuf_length * element->channels), current_srcbuf_length * element->channels * sizeof(DTYPE COMPLEX)); \
			srcbuf = gst_buffer_new_wrapped(srcbuf_data, current_srcbuf_length * element->channels * sizeof(DTYPE COMPLEX)); \
 \
			if(G_UNLIKELY(!srcbuf)) { \
				GST_ERROR_OBJECT(element, "failure creating sub-buffer"); \
				result = GST_FLOW_ERROR; \
				goto done; \
			} \
 \
			/* set flags, caps, offset, and timestamps. */ \
			GST_BUFFER_OFFSET(srcbuf) = sinkbuf_offset + element->discont_offset + offset + 1 - current_srcbuf_length; \
			GST_BUFFER_OFFSET_END(srcbuf) = GST_BUFFER_OFFSET(srcbuf) + current_srcbuf_length; \
			GST_BUFFER_PTS(srcbuf) = sinkbuf_pts + gst_util_uint64_scale_int_round(sinkbuf_dur, offset + 1 - current_srcbuf_length, length); \
			GST_BUFFER_DURATION(srcbuf) = sinkbuf_pts + gst_util_uint64_scale_int_round(sinkbuf_dur, offset + 1, length) - GST_BUFFER_PTS(srcbuf); \
			if(srcbuf_gap) \
				GST_BUFFER_FLAG_SET(srcbuf, GST_BUFFER_FLAG_GAP); \
			else \
				GST_BUFFER_FLAG_UNSET(srcbuf, GST_BUFFER_FLAG_GAP); \
 \
			/*
			 * only the first subbuffer of a buffer flagged as a
			 * discontinuity is a discontinuity.
			 */ \
			if(sinkbuf_discont && (offset + 1 - current_srcbuf_length == 0) && ((!(element->fill_discont)) || (element->last_sinkbuf_ets == 0))) \
				GST_BUFFER_FLAG_SET(srcbuf, GST_BUFFER_FLAG_DISCONT); \
			else \
				GST_BUFFER_FLAG_UNSET(srcbuf, GST_BUFFER_FLAG_DISCONT); \
			if(srcbuf_gap_next != srcbuf_gap) { \
				/* We need to reset our place in the input buffer */ \
				offset++; \
				indata += element->channels; \
				outdata += element->channels; \
				current_srcbuf_length = 1; \
				srcbuf_gap = srcbuf_gap_next; \
			} else \
				current_srcbuf_length = 0; \
 \
			/* push buffer downstream */ \
			GST_DEBUG_OBJECT(element, "pushing sub-buffer %" GST_BUFFER_BOUNDARIES_FORMAT, GST_BUFFER_BOUNDARIES_ARGS(srcbuf)); \
			result = gst_pad_push(element->srcpad, srcbuf); \
			if(G_UNLIKELY(result != GST_FLOW_OK)) { \
				GST_WARNING_OBJECT(element, "push failed: %s", gst_flow_get_name(result)); \
				goto done; \
			} \
		} \
	} \
done: \
	element->last_sinkbuf_ets = sinkbuf_pts + sinkbuf_dur; \
	element->last_sinkbuf_offset_end = sinkbuf_offset_end ? sinkbuf_offset_end : element->last_sinkbuf_offset_end; \
	return result; \
}


DEFINE_PROCESS_INBUF(guint32, )
DEFINE_PROCESS_INBUF(float, )
DEFINE_PROCESS_INBUF(double, )
DEFINE_PROCESS_INBUF(float,complex)
DEFINE_PROCESS_INBUF(double,complex)


/*
 * ============================================================================
 *
 *				     Sink Pad
 *
 * ============================================================================
 */


/*
 * sink_event()
 */


static gboolean sink_event(GstPad *pad, GstObject *parent, GstEvent *event)
{
	GSTLALInsertGap *element = GSTLAL_INSERTGAP(parent);
	gboolean success = TRUE;
	GST_DEBUG_OBJECT(pad, "Got %s event on sink pad", GST_EVENT_TYPE_NAME(event));

	switch(GST_EVENT_TYPE(event)) {
	case GST_EVENT_CAPS: {
		GstCaps *caps;
		GstAudioInfo info;
		gst_event_parse_caps(event, &caps);
		success &= gstlal_audio_info_from_caps(&info, caps);
		GstStructure *str = gst_caps_get_structure(caps, 0);
		const gchar *name = gst_structure_get_string(str, "format");
		success &= (name != NULL);
		if(success) {
			/* record stream parameters */
			element->rate = GST_AUDIO_INFO_RATE(&info);
			element->channels = GST_AUDIO_INFO_CHANNELS(&info);
			element->unit_size = GST_AUDIO_INFO_BPF(&info);
			if(!strcmp(name, GST_AUDIO_NE(U32))) {
				element->data_type = GSTLAL_INSERTGAP_U32;
				g_assert_cmpuint(element->unit_size, ==, 4);
			} else if(!strcmp(name, GST_AUDIO_NE(F32))) {
				element->data_type = GSTLAL_INSERTGAP_F32;
				g_assert_cmpuint(element->unit_size, ==, 4);
			} else if(!strcmp(name, GST_AUDIO_NE(F64))) {
				element->data_type = GSTLAL_INSERTGAP_F64;
				g_assert_cmpuint(element->unit_size, ==, 8);
			} else if(!strcmp(name, GST_AUDIO_NE(Z64))) {
				element->data_type = GSTLAL_INSERTGAP_Z64;
				g_assert_cmpuint(element->unit_size, ==, 8);
			} else if(!strcmp(name, GST_AUDIO_NE(Z128))) {
				element->data_type = GSTLAL_INSERTGAP_Z128;
				g_assert_cmpuint(element->unit_size, ==, 16);
			} else
				g_assert_not_reached();
		}
		break;
	}

	default:
		break;
	}

	if(!success) {
		gst_event_unref(event);
	} else {
		success = gst_pad_event_default(pad, parent, event);
	}
	return success;
}


/*
 * chain()
 */


static GstFlowReturn chain(GstPad *pad, GstObject *parent, GstBuffer *sinkbuf)
{
	GSTLALInsertGap *element = GSTLAL_INSERTGAP(parent);
	GstFlowReturn result = GST_FLOW_OK;
	GST_DEBUG_OBJECT(element, "received %" GST_BUFFER_BOUNDARIES_FORMAT, GST_BUFFER_BOUNDARIES_ARGS(sinkbuf));

	if(GST_BUFFER_PTS_IS_VALID(sinkbuf)) {
		/* Set the timestamp of the first output sample) */
		if(element->t0 == GST_CLOCK_TIME_NONE)
			element->t0 = GST_BUFFER_PTS(sinkbuf) + element->chop_length;

		/* If we are throwing away any initial data, do it now, and send a zero-length buffer downstream to let other elements know when to expect the first buffer */
		if(element->chop_length && GST_BUFFER_PTS(sinkbuf) + GST_BUFFER_DURATION(sinkbuf) <= element->t0) {
			GstBuffer *srcbuf = gst_buffer_new();
			GST_BUFFER_OFFSET(srcbuf) = GST_BUFFER_OFFSET(sinkbuf) + gst_util_uint64_scale_round(element->t0 - GST_BUFFER_PTS(sinkbuf), (guint64) element->rate, 1000000000);
			GST_BUFFER_OFFSET_END(srcbuf) = GST_BUFFER_OFFSET(srcbuf);
			GST_BUFFER_PTS(srcbuf) = element->t0;
			GST_BUFFER_DURATION(srcbuf) = 0;
			result = gst_pad_push(element->srcpad, srcbuf);
			gst_buffer_unref(sinkbuf);
			goto done;
		} else if(GST_BUFFER_PTS(sinkbuf) < element->t0) {
			guint64 size_removed = element->unit_size * gst_util_uint64_scale_round(element->t0 - GST_BUFFER_PTS(sinkbuf), (guint64) element->rate, 1000000000);
			guint64 time_removed = gst_util_uint64_scale_round(size_removed / element->unit_size, 1000000000, (guint64) element->rate);
			guint64 newsize = element->unit_size * (GST_BUFFER_OFFSET_END(sinkbuf) - GST_BUFFER_OFFSET(sinkbuf)) - size_removed;
			gst_buffer_resize(sinkbuf, size_removed, newsize);
			GST_BUFFER_OFFSET(sinkbuf) = GST_BUFFER_OFFSET(sinkbuf) + size_removed / element->unit_size;
			GST_BUFFER_PTS(sinkbuf) = GST_BUFFER_PTS(sinkbuf) + time_removed;
			GST_BUFFER_DURATION(sinkbuf) = GST_BUFFER_DURATION(sinkbuf) - time_removed;
		}
	}

	/* if buffer does not possess valid metadata or is zero length and we are not filling in discontinuities, push gap downstream */
	if(!(GST_BUFFER_PTS_IS_VALID(sinkbuf) && GST_BUFFER_DURATION_IS_VALID(sinkbuf) && GST_BUFFER_OFFSET_IS_VALID(sinkbuf) && GST_BUFFER_OFFSET_END_IS_VALID(sinkbuf)) || (!element->fill_discont && (GST_BUFFER_DURATION(sinkbuf) == 0 || GST_BUFFER_OFFSET(sinkbuf) == GST_BUFFER_OFFSET_END(sinkbuf)))) {
		GST_DEBUG_OBJECT(element, "pushing gap buffer at timestamp %lu seconds", (long unsigned) GST_TIME_AS_SECONDS(GST_BUFFER_PTS(sinkbuf)));
		GstBuffer *srcbuf;
		srcbuf = gst_buffer_copy(sinkbuf);
		GST_BUFFER_FLAG_SET(srcbuf, GST_BUFFER_FLAG_GAP);
		result = gst_pad_push(element->srcpad, srcbuf);
		if(G_UNLIKELY(result != GST_FLOW_OK))
			GST_WARNING_OBJECT(element, "push failed: %s", gst_flow_get_name(result));
		goto done;
	}

	/* if buffer is zero length and we are filling in discontinuities, fill it in, unless it has no valid timestamp. */
	if(element->fill_discont && (GST_BUFFER_DURATION(sinkbuf) == 0 || GST_BUFFER_OFFSET(sinkbuf) == GST_BUFFER_OFFSET_END(sinkbuf))) {
		if(GST_BUFFER_PTS_IS_VALID(sinkbuf) && GST_BUFFER_PTS(sinkbuf) > element->last_sinkbuf_ets) {
			switch(element->data_type) {
			case GSTLAL_INSERTGAP_U32:
				result = process_inbuf_guint32(NULL, NULL, element, TRUE, TRUE, 0, 0, 0, GST_BUFFER_PTS(sinkbuf), FALSE);
				break;
			case GSTLAL_INSERTGAP_F32:
				result = process_inbuf_float(NULL, NULL, element, TRUE, TRUE, 0, 0, 0, GST_BUFFER_PTS(sinkbuf), FALSE);
				break;
			case GSTLAL_INSERTGAP_F64:
				result = process_inbuf_double(NULL, NULL, element, TRUE, TRUE, 0, 0, 0, GST_BUFFER_PTS(sinkbuf), FALSE);
				break;
			case GSTLAL_INSERTGAP_Z64:
				result = process_inbuf_floatcomplex(NULL, NULL, element, TRUE, TRUE, 0, 0, 0, GST_BUFFER_PTS(sinkbuf), TRUE);
				break;
			case GSTLAL_INSERTGAP_Z128:
				result = process_inbuf_doublecomplex(NULL, NULL, element, TRUE, TRUE, 0, 0, 0, GST_BUFFER_PTS(sinkbuf), TRUE);
				break;
			default:
				g_assert_not_reached();
			}
		} else {
			GST_DEBUG_OBJECT(element, "dropping zero length buffer at timestamp %lu seconds", (long unsigned) GST_TIME_AS_SECONDS(GST_BUFFER_PTS(sinkbuf)));
			gst_buffer_unref(sinkbuf);
		}
		goto done;
	}

	GstMapInfo inmap;
	gst_buffer_map(sinkbuf, &inmap, GST_MAP_READ);

	/* We'll need these to decide gaps, offsets, and timestamps on the outgoing buffer(s) */
	gboolean sinkbuf_gap = GST_BUFFER_FLAG_IS_SET(sinkbuf, GST_BUFFER_FLAG_GAP);
	gboolean sinkbuf_discont = GST_BUFFER_FLAG_IS_SET(sinkbuf, GST_BUFFER_FLAG_DISCONT);
	guint64 sinkbuf_offset = GST_BUFFER_OFFSET(sinkbuf);
	guint64 sinkbuf_offset_end = GST_BUFFER_OFFSET_END(sinkbuf);
	GstClockTime sinkbuf_dur = GST_BUFFER_DURATION(sinkbuf);
	GstClockTime sinkbuf_pts = GST_BUFFER_PTS(sinkbuf);

	g_assert_cmpuint(inmap.size % element->unit_size, ==, 0);
	if(!element->chop_length || sinkbuf_pts > element->t0)
		g_assert_cmpuint((sinkbuf_offset_end - sinkbuf_offset), ==, inmap.size / element->unit_size); /* sanity checks */

	/* outdata will be filled with the data that goes on the outgoing buffer(s) */
	void *outdata;
	outdata = g_malloc((sinkbuf_offset_end - sinkbuf_offset) * element->unit_size);

	switch(element->data_type) {
	case GSTLAL_INSERTGAP_U32:
		result = process_inbuf_guint32((guint32 *) inmap.data, outdata, element, sinkbuf_gap, sinkbuf_discont, sinkbuf_offset, sinkbuf_offset_end, sinkbuf_dur, sinkbuf_pts, FALSE);
		break;
	case GSTLAL_INSERTGAP_F32:
		result = process_inbuf_float((float *) inmap.data, outdata, element, sinkbuf_gap, sinkbuf_discont, sinkbuf_offset, sinkbuf_offset_end, sinkbuf_dur, sinkbuf_pts, FALSE);
		break;
	case GSTLAL_INSERTGAP_F64:
		result = process_inbuf_double((double *) inmap.data, outdata, element, sinkbuf_gap, sinkbuf_discont, sinkbuf_offset, sinkbuf_offset_end, sinkbuf_dur, sinkbuf_pts, FALSE);
		break;
	case GSTLAL_INSERTGAP_Z64:
		result = process_inbuf_floatcomplex((float complex *) inmap.data, outdata, element, sinkbuf_gap, sinkbuf_discont, sinkbuf_offset, sinkbuf_offset_end, sinkbuf_dur, sinkbuf_pts, TRUE);
		break;
	case GSTLAL_INSERTGAP_Z128:
		result = process_inbuf_doublecomplex((double complex *) inmap.data, outdata, element, sinkbuf_gap, sinkbuf_discont, sinkbuf_offset, sinkbuf_offset_end, sinkbuf_dur, sinkbuf_pts, TRUE);
		break;

	default:
		g_assert_not_reached();
	}

	g_free(outdata);
	outdata = NULL;
	gst_buffer_unmap(sinkbuf, &inmap);
	gst_buffer_unref(sinkbuf);
	/*
	 * done
	 */

done:
	return result;
}


/*
 * ============================================================================
 *
 *			  GObject Method Overrides
 *
 * ============================================================================
 */


/*
 * properties
 */


enum property {
	ARG_INSERT_GAP = 1,
	ARG_REMOVE_GAP,
	ARG_REMOVE_NAN,
	ARG_REMOVE_INF,
	ARG_FILL_DISCONT,
	ARG_REPLACE_VALUE,
	ARG_BAD_DATA_INTERVALS,
	ARG_BLOCK_DURATION,
	ARG_CHOP_LENGTH
};


static void set_property(GObject *object, enum property prop_id, const GValue *value, GParamSpec *pspec)
{
	GSTLALInsertGap *element = GSTLAL_INSERTGAP(object);

	GST_OBJECT_LOCK(element);

	switch (prop_id) {
	case ARG_INSERT_GAP:
		element->insert_gap = g_value_get_boolean(value);
		break;
	case ARG_REMOVE_GAP:
		element->remove_gap = g_value_get_boolean(value);
		break;
	case ARG_REMOVE_NAN:
		element->remove_nan = g_value_get_boolean(value);
		break;
	case ARG_REMOVE_INF:
		element->remove_inf = g_value_get_boolean(value);
		break;
	case ARG_FILL_DISCONT:
		element->fill_discont = g_value_get_boolean(value);
		break;
	case ARG_REPLACE_VALUE:
		element->replace_value = g_value_get_double(value);
		break;
	case ARG_BAD_DATA_INTERVALS: {
		if(element->bad_data_intervals) {
			g_free(element->bad_data_intervals);
			element->bad_data_intervals = NULL;
		}
		element->bad_data_intervals = gstlal_doubles_from_g_value_array(g_value_get_boxed(value), NULL, &element->array_length);
		if(element->array_length % 2)
			GST_ERROR_OBJECT(element, "Array length for property bad_data_intervals must be even");
		break;
	}
	case ARG_BLOCK_DURATION:
		element->block_duration = g_value_get_uint64(value);
		break;
	case ARG_CHOP_LENGTH:
		element->chop_length = g_value_get_uint64(value);
		break;
	default:
		G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
		break;
	}

	GST_OBJECT_UNLOCK(element);
}


static void get_property(GObject *object, enum property prop_id, GValue *value, GParamSpec *pspec)
{
	GSTLALInsertGap *element = GSTLAL_INSERTGAP(object);

	GST_OBJECT_LOCK(element);

	switch (prop_id) {
	case ARG_INSERT_GAP:
		g_value_set_boolean(value, element->insert_gap);
		break;
	case ARG_REMOVE_GAP:
		g_value_set_boolean(value, element->remove_gap);
		break;
	case ARG_REMOVE_NAN:
		g_value_set_boolean(value, element->remove_nan);
		break;
	case ARG_REMOVE_INF:
		g_value_set_boolean(value, element->remove_inf);
		break;
	case ARG_FILL_DISCONT:
		g_value_set_boolean(value, element->fill_discont);
		break;
	case ARG_REPLACE_VALUE:
		g_value_set_double(value, element->replace_value);
		break;
	case ARG_BAD_DATA_INTERVALS:
		g_value_take_boxed(value, gstlal_g_value_array_from_doubles(element->bad_data_intervals, element->array_length));
		break;
	case ARG_BLOCK_DURATION:
		g_value_set_uint64(value, element->block_duration);
		break;
	case ARG_CHOP_LENGTH:
		g_value_set_uint64(value, element->chop_length);
		break;
	default:
		G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
		break;
	}

	GST_OBJECT_UNLOCK(element);
}


/*
 * finalize()
 */


static void finalize(GObject *object)
{
	GSTLALInsertGap *element = GSTLAL_INSERTGAP(object);

	gst_object_unref(element->sinkpad);
	element->sinkpad = NULL;
	gst_object_unref(element->srcpad);
	element->srcpad = NULL;
	g_free(element->bad_data_intervals);
	element->bad_data_intervals = NULL;

	G_OBJECT_CLASS(gstlal_insertgap_parent_class)->finalize(object);
}


/*
 * class_init()
 */


static void gstlal_insertgap_class_init(GSTLALInsertGapClass *klass)
{
	GObjectClass *gobject_class = G_OBJECT_CLASS(klass);
	GstElementClass *element_class = GST_ELEMENT_CLASS(klass);

	gst_element_class_set_details_simple(
		element_class,
		"Replace unwanted data with gaps",
		"Filter",
		"Replace unwanted data, specified with the property bad-data-intervals, with gaps.\n\t\t\t   "
		"Also can replace with another value, given by replace-value. Can also remove gaps\n\t\t\t   "
		"where data is acceptable, and fill in discontinuities if desired.",
		"Aaron Viets <aaron.viets@ligo.org>"
	);

	gobject_class->set_property = GST_DEBUG_FUNCPTR(set_property);
	gobject_class->get_property = GST_DEBUG_FUNCPTR(get_property);
	gobject_class->finalize = GST_DEBUG_FUNCPTR(finalize);

	gst_element_class_add_pad_template(element_class, gst_static_pad_template_get(&src_factory));
	gst_element_class_add_pad_template(element_class, gst_static_pad_template_get(&sink_factory));

	g_object_class_install_property(
		gobject_class,
		ARG_INSERT_GAP,
		g_param_spec_boolean(
			"insert-gap",
			"Insert gap",
			"If set to true (default), any data fitting the criteria specified by the property\n\t\t\t"
			"bad-data-intervals is replaced with gaps. Also, NaN's and inf's are replaced with\n\t\t\t"
			"gaps if the properties remove-nan and remove-inf are set to true, respectively.",
			TRUE,
			G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT
		)
	);

	g_object_class_install_property(
		gobject_class,
		ARG_REMOVE_GAP,
		g_param_spec_boolean(
			"remove-gap",
			"Remove gap",
			"If set to true, any data in an input gap buffer that does not fit the criteria\n\t\t\t"
			"specified by the property bad-data-intervals will be marked as non-gap. If the\n\t\t\t"
			"property insert-gap is false and remove-gap is true, gaps with unacceptable\n\t\t\t"
			"data will be replaced by the value specified by the property replace-value.",
			FALSE,
			G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT
		)
	);

	g_object_class_install_property(
		gobject_class,
		ARG_REMOVE_NAN,
		g_param_spec_boolean(
			"remove-nan",
			"Remove NaN",
			"If set to true (default), NaN's in the data stream will be replaced with gaps\n\t\t\t"
			"and/or the replace-value, as specified by user.",
			TRUE,
			G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT
		)
	);

	g_object_class_install_property(
		gobject_class,
		ARG_REMOVE_INF,
		g_param_spec_boolean(
			"remove-inf",
			"Remove inf",
			"If set to true (default), infinities in the data stream will be replaced with\n\t\t\t"
			"gaps and/or the replace-value, as specified by user.",
			TRUE,
			G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT
		)
	);

	g_object_class_install_property(
		gobject_class,
		ARG_FILL_DISCONT,
		g_param_spec_boolean(
			"fill-discont",
			"Fill discontinuity",
			"If set to true, discontinuities in the data stream will be filled with the\n\t\t\t"
			"replace-value (if set, otherwise 0), and gapped if insert-gap is true.",
			FALSE,
			G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT
		)
	);

	g_object_class_install_property(
		gobject_class,
		ARG_REPLACE_VALUE,
		g_param_spec_double(
			"replace-value",
			"Replace value",
			"If set, this value is used to replace any data that fits the criteria\n\t\t\t"
			"specified by the property bad-data-intervals. If unset, values are not replaced.", 
			-G_MAXDOUBLE, G_MAXDOUBLE, G_MAXDOUBLE,
			G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT
		)
	);

	g_object_class_install_property(
		gobject_class,
		ARG_BAD_DATA_INTERVALS,
		g_param_spec_value_array(
			"bad-data-intervals",
			"Bad data intervals",
			"Array containing minima and maxima of closed intervals in which data is\n\t\t\t"
			"considered unacceptable and will be replaced with gaps and/or the replace-value.\n\t\t\t"
			"Array indices 0, 2, 4, etc., represent maxima, and array indices 1, 3, 5, etc.,\n\t\t\t"
			"represent the corresponding minima.",
			g_param_spec_double(
				"coefficient",
				"Coefficient",
				"Coefficient",
				-G_MAXDOUBLE, G_MAXDOUBLE, 0.0,
				G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS
			),
			G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS
		)
	);

	g_object_class_install_property(
		gobject_class,
		ARG_BLOCK_DURATION,
		g_param_spec_uint64(
			"block-duration",
			"Block duration",
			"Maximum output buffer duration in nanoseconds. Buffers may be smaller than this.\n\t\t\t"
			"Default is to not change buffer length except as required by added/removed gaps.",
			0, G_MAXUINT64, G_MAXUINT64 / 2,
			G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT
		)
	);

	g_object_class_install_property(
		gobject_class,
		ARG_CHOP_LENGTH,
		g_param_spec_uint64(
			"chop-length",
			"Chop length",
			"Amount of initial data to throw away before producing output data, in nanoseconds.",
			0, G_MAXUINT64, 0,
			G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT
		)
	);
}


/*
 * instance init
 */


static void gstlal_insertgap_init(GSTLALInsertGap *element)
{
	GstPad *pad;

	gst_element_create_all_pads(GST_ELEMENT(element));

	/* configure (and ref) sink pad */
	pad = gst_element_get_static_pad(GST_ELEMENT(element), "sink");
	gst_pad_set_event_function(pad, GST_DEBUG_FUNCPTR(sink_event));
	gst_pad_set_chain_function(pad, GST_DEBUG_FUNCPTR(chain));
	GST_PAD_SET_PROXY_CAPS(pad);
	GST_PAD_SET_PROXY_ALLOCATION(pad);
	GST_PAD_SET_PROXY_SCHEDULING(pad);
	element->sinkpad = pad;

	/* retrieve (and ref) src pad */
	pad = gst_element_get_static_pad(GST_ELEMENT(element), "src");
	GST_PAD_SET_PROXY_CAPS(pad);
	GST_PAD_SET_PROXY_ALLOCATION(pad);
	GST_PAD_SET_PROXY_SCHEDULING(pad);
	element->srcpad = pad;

	/* internal data */
	element->t0 = GST_CLOCK_TIME_NONE;
	element->bad_data_intervals = NULL;
	element->array_length = 0;
	element->rate = 0;
	element->channels = 0;
	element->unit_size = 0;
	element->last_sinkbuf_ets = 0;
	element->last_sinkbuf_offset_end = 0;
	element->discont_offset = 0;
	element->discont_time = 0;
	element->empty_bufs = 0;
}