1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
#!/usr/bin/env python2
# -*- coding: utf-8 -*-
# @file
# @author Pieter Robyns <pieter.robyns@uhasselt.be>
# @section LICENSE
#
# Gr-gsm 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 3, or (at your option)
# any later version.
#
# Gr-gsm 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 gr-gsm; see the file COPYING. If not, write to
# the Free Software Foundation, Inc., 51 Franklin Street,
# Boston, MA 02110-1301, USA.
#
#
##################################################
# gr-gsm channelizer
#
# Standalone application to channelize a wideband
# GSM capture file into multiple seperate capture
# files for the specified ARFCNs.
##################################################
from gnuradio import blocks
from gnuradio import eng_notation
from gnuradio import filter
from gnuradio import gr
from gnuradio.eng_option import eng_option
from gnuradio.filter import firdes
from argparse import ArgumentParser, ArgumentTypeError, RawDescriptionHelpFormatter
import grgsm.arfcn as arfcn
from gnuradio.filter import pfb
import math
import os
EXTRA_HELP = """
Example usage:
grgsm_channelize.py -c my_wideband_capture.cfile -f 925.2e6 990 991 992 993 994 995 1019 1020 1021 1022 1023
The above example will channelize my_wideband_capture.cfile, in this case a cfile captured at
925.2 MHz centered (ARFCN 975) and 2 Msps. As a result, 12 files will be generated for
ARFCNs 975 - 1023 at 1 Msps each.
"""
def eng_float(value):
try:
return eng_notation.str_to_num(value)
except:
raise ArgumentTypeError("invalid engineering notation value: {0}".format(value))
def gsm_band(value):
choices = arfcn.get_bands()
if value in choices:
return value
else:
raise ArgumentTypeError("invalid GSM band: {0}. Possible choices are: {1}".format(value, choices))
class grgsm_channelize(gr.top_block):
def __init__(self, channels, resamp_rate, fc, band, samp_rate, input_file, dest_dir, data_type="complex"):
gr.top_block.__init__(self, "grgsm_channelize")
##################################################
# Parameters
##################################################
self.channels = channels
self.resamp_rate = resamp_rate
self.fc = fc
self.band = band
self.samp_rate = samp_rate
self.blocks_resamplers = {}
self.blocks_rotators = {}
self.blocks_file_sinks = {}
##################################################
# Blocks and connections
##################################################
self.source = None
if data_type == "ishort":
self.blocks_file_source = blocks.file_source(gr.sizeof_short, input_file, False)
self.source = blocks.interleaved_short_to_complex(False, False)
self.connect((self.blocks_file_source, 0), (self.source, 0))
elif data_type == "complex":
self.source = blocks.file_source(gr.sizeof_gr_complex, input_file, False)
fc_str = eng_notation.num_to_str(fc)
print("Extracting channels %s, given that the center frequency is at %s" % (str(channels), eng_notation.num_to_str(fc)))
for channel in channels:
channel_freq = arfcn.arfcn2downlink(channel, band)
if channel_freq is None:
print("Warning: invalid ARFCN %d for band %s" % (channel, band))
continue
freq_diff = channel_freq - fc
freq_diff_str = "+" if 0 <= freq_diff else ""
freq_diff_str += eng_notation.num_to_str(freq_diff)
print("ARFCN %d is at %sHz %sHz" % (channel, fc_str, freq_diff_str))
self.blocks_resamplers[channel] = pfb.arb_resampler_ccf( resamp_rate, taps=None, flt_size=32)
self.blocks_rotators[channel] = blocks.rotator_cc(-2*math.pi*(freq_diff)/samp_rate)
self.connect( (self.source, 0), (self.blocks_rotators[channel], 0) )
self.connect( (self.blocks_rotators[channel], 0), (self.blocks_resamplers[channel], 0) )
self.blocks_file_sinks[channel] = blocks.file_sink(gr.sizeof_gr_complex, dest_dir+"/out_" + str(channel) + ".cfile", False)
self.blocks_file_sinks[channel].set_unbuffered(False)
self.connect((self.blocks_resamplers[channel], 0), (self.blocks_file_sinks[channel], 0))
if __name__ == '__main__':
parser = ArgumentParser(formatter_class=RawDescriptionHelpFormatter, description='Split wideband a GSM capture into seperate files per ARFCN.', add_help=True, epilog=EXTRA_HELP)
parser.add_argument(dest="channel", type=int, nargs='+',
help="List of ARFCNs")
parser.add_argument("-s", "--samp-rate", dest="samp_rate", type=eng_float, default=eng_notation.num_to_str(2e6),
help="Sample rate of the wideband capture file [default=%(default)s]")
parser.add_argument("-f", "--fc", dest="fc", type=eng_float, default=eng_notation.num_to_str(935e6), required=True,
help="Carrier frequency in Hz [default=%(default)s]")
parser.add_argument("-b", "--band", dest="band", type=gsm_band, default='E-GSM',
help="GSM band [default=%(default)s]") #TODO: add automatic discovery based on fc
parser.add_argument("-o", "--out-samp-rate", dest="out_samp_rate", type=eng_float, default=eng_notation.num_to_str(1e6),
help="Sample rate of the output capture files [default=%(default)s]")
parser.add_argument("-i", "--input_file", dest="input_file", type=str, required=True,
help="Path to wideband GSM capture file")
parser.add_argument("-t", "--data_type", dest="data_type", type=str, choices=["complex","ishort"], default="complex",
help="Type of the input file [default=%(default)s]")
parser.add_argument("-d", "--dest_dir", dest="dest_dir", type=str,
help="Destination directory - if not given defaults to input file name without extension")
args = parser.parse_args()
if not os.path.exists(args.input_file):
raise IOError(args.input_file + " does not exist")
input_filename, _ = os.path.splitext(os.path.basename(args.input_file))
if args.dest_dir is None:
args.dest_dir = input_filename
if not os.path.exists("./"+args.dest_dir):
os.makedirs("./"+args.dest_dir)
if args.samp_rate % args.out_samp_rate != 0:
raise Exception("Input sample rate should be multiple of output sample rate in order to get integer decimation.")
resamp_rate = args.out_samp_rate / args.samp_rate
print("Input sample rate: " + eng_notation.num_to_str(args.samp_rate))
print("Output sample rate: " + eng_notation.num_to_str(args.out_samp_rate))
print("==> using resample rate of " + str(resamp_rate))
tb = grgsm_channelize(channels=args.channel,
resamp_rate=resamp_rate,
fc=args.fc,
band=args.band,
samp_rate=args.samp_rate,
input_file=args.input_file,
dest_dir=args.dest_dir,
data_type=args.data_type
)
tb.start()
tb.wait()
print("Done!")
|
