gstlal 1.13.0
Loading...
Searching...
No Matches
resample_test_01.py
1#!/usr/bin/env python3
2#
3# =============================================================================
4#
5# Preamble
6#
7# =============================================================================
8#
9
10
11import numpy
12import sys
13from gstlal import pipeparts
14import test_common
15
16
17#
18# =============================================================================
19#
20# Pipelines
21#
22# =============================================================================
23#
24
25
26#
27# check for timestamp drift in the audioresample
28#
29
30
31def resample_test_01a(pipeline, name):
32 #
33 # try changing these. test should still work!
34 #
35
36 in_rate = 2048 # Hz
37 out_rate = 512 # Hz
38 quality = 9 # [0, 9]
39 gap_frequency = None # Hz
40 gap_threshold = 0.0 # of 1
41 buffer_length = 1.0 # seconds
42 test_duration = 20.0 # seconds
43
44 #
45 # build pipeline
46 #
47
48 head = test_common.gapped_test_src(pipeline, buffer_length = buffer_length, rate = in_rate, test_duration = test_duration, gap_frequency = gap_frequency, gap_threshold = gap_threshold, control_dump_filename = "%s_control.dump" % name)
49 elem = pipeparts.Gst.ElementFactory.make("audiocheblimit", None)
50 elem.set_property("mode", 0)
51 elem.set_property("cutoff", .95 * out_rate / 2.0)
52 pipeline.add(elem)
53 head.link(elem)
54 head = elem
55 head = tee = pipeparts.mktee(pipeline, head)
56
57 head = pipeparts.mkcapsfilter(pipeline, pipeparts.mkresample(pipeline, head, quality = quality), "audio/x-raw, rate=%d" % out_rate)
58 head = pipeparts.mkchecktimestamps(pipeline, head)
59 pipeparts.mknxydumpsink(pipeline, pipeparts.mkqueue(pipeline, head), "%s_out.dump" % name)
60 pipeparts.mknxydumpsink(pipeline, pipeparts.mkqueue(pipeline, tee), "%s_in.dump" % name)
61
62 #
63 # done
64 #
65
66 return pipeline
67
68
69#
70# check for timestamp drift in the audioresample
71#
72
73
74def resample_test_01b(pipeline, name):
75 #
76 # try changing these. test should still work!
77 #
78
79 in_rate = 512 # Hz
80 out_rate = 2048 # Hz
81 quality = 9 # [0, 9]
82 gap_frequency = None # Hz
83 gap_threshold = 0.0 # of 1
84 buffer_length = 1.0 # seconds
85 test_duration = 20.0 # seconds
86
87 #
88 # build pipeline
89 #
90
91 head = test_common.gapped_test_src(pipeline, buffer_length = buffer_length, rate = in_rate, test_duration = test_duration, gap_frequency = gap_frequency, gap_threshold = gap_threshold, control_dump_filename = "%s_control.dump" % name)
92 elem = pipeparts.Gst.ElementFactory.make("audiocheblimit", None)
93 elem.set_property("mode", 0)
94 elem.set_property("cutoff", .95 * out_rate / 2.0)
95 pipeline.add(elem)
96 head.link(elem)
97 head = elem
98 head = tee = pipeparts.mktee(pipeline, head)
99
100 head = pipeparts.mkcapsfilter(pipeline, pipeparts.mkresample(pipeline, head, quality = quality), "audio/x-raw, rate=%d" % out_rate)
101 head = pipeparts.mkchecktimestamps(pipeline, head)
102 head = pipeparts.mkcapsfilter(pipeline, pipeparts.mkresample(pipeline, head, quality = quality), "audio/x-raw, rate=%d" % in_rate)
103 head = pipeparts.mkchecktimestamps(pipeline, head)
104 head = pipeparts.mkcapsfilter(pipeline, pipeparts.mkresample(pipeline, head, quality = quality), "audio/x-raw, rate=%d" % out_rate)
105 head = pipeparts.mkchecktimestamps(pipeline, head)
106 head = pipeparts.mkcapsfilter(pipeline, pipeparts.mkresample(pipeline, head, quality = quality), "audio/x-raw, rate=%d" % in_rate)
107 head = pipeparts.mkchecktimestamps(pipeline, head)
108 pipeparts.mknxydumpsink(pipeline, pipeparts.mkqueue(pipeline, head), "%s_out.dump" % name)
109 pipeparts.mknxydumpsink(pipeline, pipeparts.mkqueue(pipeline, tee), "%s_in.dump" % name)
110
111 #
112 # done
113 #
114
115 return pipeline
116
117
118#
119# =============================================================================
120#
121# Main
122#
123# =============================================================================
124#
125
126
127test_common.build_and_run(resample_test_01a, "resample_test_01a")
128test_common.build_and_run(resample_test_01b, "resample_test_01b")
129
build_and_run(pipelinefunc, name, segment=None, **pipelinefunc_kwargs)
gapped_test_src(pipeline, buffer_length=1.0, rate=2048, width=64, channels=1, test_duration=10.0, wave=5, freq=0, gap_frequency=None, gap_threshold=None, control_dump_filename=None, tags=None, is_live=False, verbose=True)