How to use the metpy.io.nexrad.NamedStruct function in MetPy
To help you get started, we’ve selected a few MetPy examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Unidata / MetPy / metpy / io / nexrad.py View on Github 
def _unpack_packet_radial_data(self, data):
hdr_fmt = NamedStruct([('ind_first_bin', 'H'), ('nbins', 'H'),
('i_center', 'h'), ('j_center', 'h'), ('scale_factor', 'h'),
('num_rad', 'H')], '>', 'RadialHeader')
rad_fmt = NamedStruct([('num_hwords', 'H'), ('start_angle', 'h'),
('angle_delta', 'h')], '>', 'RadialData')
hdr = hdr_fmt.unpack_from(data)
print hdr
size = hdr_fmt.size
rads = []
for i in range(hdr.num_rad):
rad = rad_fmt.unpack_from(data, size)
size += rad_fmt.size
start_az = rad.start_angle * 0.1
end_az = start_az + rad.angle_delta * 0.1
rad_data = data[size:size + rad.num_hwords * 2]
# Unpack Run-length encoded datadef _read_data(self):
ctm_hdr = FileStruct('12c', self._fobj)
ctm_hdr.read()
msg_hdr_fields = [('size_hw', 'H'), ('rda_channel', 'B'),
('msg_type', 'B'), ('seq_num', 'H'),
('date', 'H'), ('time_ms', 'I'),
('num_segments', 'H'), ('segment_num', 'H')]
msg_hdr_fmt = NamedStruct(msg_hdr_fields, '>', 'MsgHdr')
while True:
msg_hdr = msg_hdr_fmt.unpack_file(self._fobj)
if msg_hdr.msg_type in (1,31):
dt = nexrad_to_datetime(msg_hdr.date, msg_hdr.time_ms)
print msg_hdr, str(dt)
msg = self._fobj.read(2 * msg_hdr.size_hw)
try:
getattr(self, '_decode_msg%d' % msg_hdr.msg_type)(msg)
except AttributeError:
pass
# print 'Unknown Message Type: %d' % msg_hdr.msg_type
if msg_hdr.msg_type != 31:
self._fobj.read(self.AR2_BLOCKSIZE - len(msg))def _unpack_packet_radial_data(self, data):
hdr_fmt = NamedStruct([('ind_first_bin', 'H'), ('nbins', 'H'),
('i_center', 'h'), ('j_center', 'h'), ('scale_factor', 'h'),
('num_rad', 'H')], '>', 'RadialHeader')
rad_fmt = NamedStruct([('num_hwords', 'H'), ('start_angle', 'h'),
('angle_delta', 'h')], '>', 'RadialData')
hdr = hdr_fmt.unpack_from(data)
print hdr
size = hdr_fmt.size
rads = []
for i in range(hdr.num_rad):
rad = rad_fmt.unpack_from(data, size)
size += rad_fmt.size
start_az = rad.start_angle * 0.1
end_az = start_az + rad.angle_delta * 0.1
rad_data = data[size:size + rad.num_hwords * 2]
# Unpack Run-length encoded data
unpacked = []
for run in map(ord, rad_data):
num,val = run>>4, run&0x0F('msg_len', 'L'), ('src_id', 'h'), ('dest_id', 'h'),
('num_blks', 'H')], '>', 'MsgHdr')
prod_desc_fmt = NamedStruct([('divider', 'h'), ('lat', 'l'), ('lon', 'l'),
('height', 'h'), ('prod_code', 'h'), ('op_mode', 'h'),
('vcp', 'h'), ('seq_num', 'H'), ('vol_num', 'H'),
('vol_date', 'H'), ('vol_start_time', 'l'), ('prod_gen_date', 'H'),
('prod_gen_time', 'l'), ('dep1', 'h'), ('dep2', 'h'), ('el_num', 'H'),
('dep3', 'h'), ('thr1', 'H'), ('thr2', 'H'), ('thr3', 'H'),
('thr4', 'H'), ('thr5', 'H'), ('thr6', 'H'), ('thr7', 'H'),
('thr8', 'H'), ('thr9', 'H'), ('thr10', 'H'), ('thr11', 'H'),
('thr12', 'H'), ('thr13', 'H'), ('thr14', 'H'), ('thr15', 'H'),
('thr16', 'H'), ('dep4', 'h'), ('dep5', 'h'), ('dep6', 'h'),
('dep7', 'h'), ('dep8', 'h'), ('dep9', 'h'), ('dep10', 'h'),
('n_maps', 'h'), ('sym_off', 'L'), ('graph_off', 'L'),
('tab_off', 'L')], '>', 'ProdDesc')
sym_block_fmt = NamedStruct([('divider', 'h'), ('block_id', 'h'),
('block_len', 'L'), ('nlayer', 'H')], '>', 'SymBlock')
sym_layer_fmt = NamedStruct([('divider', 'h'), ('length', 'L')], '>',
'SymLayer')
graph_block_fmt = NamedStruct([('divider', 'h'), ('block_id', 'h'),
('block_len', 'L'), ('npages', 'H'), ('pagenum', 'H'),
('page_len', 'H')], '>', 'GraphBlock')
def __init__(self, fname):
self.packet_map = {0xaf1f:self._unpack_packet_radial_data}
self._fobj = open(fname, 'rb')
# Read off the UNISYS header, the first 30 bytes
self._start_offset = 30
self._fobj.seek(self._start_offset)
# Set up places to store data and metadataprod_desc_fmt = NamedStruct([('divider', 'h'), ('lat', 'l'), ('lon', 'l'),
('height', 'h'), ('prod_code', 'h'), ('op_mode', 'h'),
('vcp', 'h'), ('seq_num', 'H'), ('vol_num', 'H'),
('vol_date', 'H'), ('vol_start_time', 'l'), ('prod_gen_date', 'H'),
('prod_gen_time', 'l'), ('dep1', 'h'), ('dep2', 'h'), ('el_num', 'H'),
('dep3', 'h'), ('thr1', 'H'), ('thr2', 'H'), ('thr3', 'H'),
('thr4', 'H'), ('thr5', 'H'), ('thr6', 'H'), ('thr7', 'H'),
('thr8', 'H'), ('thr9', 'H'), ('thr10', 'H'), ('thr11', 'H'),
('thr12', 'H'), ('thr13', 'H'), ('thr14', 'H'), ('thr15', 'H'),
('thr16', 'H'), ('dep4', 'h'), ('dep5', 'h'), ('dep6', 'h'),
('dep7', 'h'), ('dep8', 'h'), ('dep9', 'h'), ('dep10', 'h'),
('n_maps', 'h'), ('sym_off', 'L'), ('graph_off', 'L'),
('tab_off', 'L')], '>', 'ProdDesc')
sym_block_fmt = NamedStruct([('divider', 'h'), ('block_id', 'h'),
('block_len', 'L'), ('nlayer', 'H')], '>', 'SymBlock')
sym_layer_fmt = NamedStruct([('divider', 'h'), ('length', 'L')], '>',
'SymLayer')
graph_block_fmt = NamedStruct([('divider', 'h'), ('block_id', 'h'),
('block_len', 'L'), ('npages', 'H'), ('pagenum', 'H'),
('page_len', 'H')], '>', 'GraphBlock')
def __init__(self, fname):
self.packet_map = {0xaf1f:self._unpack_packet_radial_data}
self._fobj = open(fname, 'rb')
# Read off the UNISYS header, the first 30 bytes
self._start_offset = 30
self._fobj.seek(self._start_offset)
# Set up places to store data and metadata
# self.data = []
# self.metadata = dict()data_hdr.vol_const_ptr)
print vol_consts
el_consts = self.msg31_el_const_fmt.unpack_from(msg,
data_hdr.el_const_ptr)
print el_consts
rad_consts = self.rad_const_fmt.unpack_from(msg, data_hdr.rad_const_ptr)
print rad_consts
def _decode_msg1(self, msg):
pass
class Level3File(object):
header_fmt = NamedStruct([('code', 'H'), ('date', 'H'), ('time', 'l'),
('msg_len', 'L'), ('src_id', 'h'), ('dest_id', 'h'),
('num_blks', 'H')], '>', 'MsgHdr')
prod_desc_fmt = NamedStruct([('divider', 'h'), ('lat', 'l'), ('lon', 'l'),
('height', 'h'), ('prod_code', 'h'), ('op_mode', 'h'),
('vcp', 'h'), ('seq_num', 'H'), ('vol_num', 'H'),
('vol_date', 'H'), ('vol_start_time', 'l'), ('prod_gen_date', 'H'),
('prod_gen_time', 'l'), ('dep1', 'h'), ('dep2', 'h'), ('el_num', 'H'),
('dep3', 'h'), ('thr1', 'H'), ('thr2', 'H'), ('thr3', 'H'),
('thr4', 'H'), ('thr5', 'H'), ('thr6', 'H'), ('thr7', 'H'),
('thr8', 'H'), ('thr9', 'H'), ('thr10', 'H'), ('thr11', 'H'),
('thr12', 'H'), ('thr13', 'H'), ('thr14', 'H'), ('thr15', 'H'),
('thr16', 'H'), ('dep4', 'h'), ('dep5', 'h'), ('dep6', 'h'),
('dep7', 'h'), ('dep8', 'h'), ('dep9', 'h'), ('dep10', 'h'),
('n_maps', 'h'), ('sym_off', 'L'), ('graph_off', 'L'),
('tab_off', 'L')], '>', 'ProdDesc')
sym_block_fmt = NamedStruct([('divider', 'h'), ('block_id', 'h'),
('block_len', 'L'), ('nlayer', 'H')], '>', 'SymBlock')
sym_layer_fmt = NamedStruct([('divider', 'h'), ('length', 'L')], '>',
'SymLayer')def _read_volume_header(self):
vol_fields = [('version', '9s'), ('vol_num', '3s'), ('date', 'L'),
('time_ms', 'L'), ('stid', '4s')]
vol_hdr_fmt = NamedStruct(vol_fields, '>', 'VolHdr')
vol_hdr = vol_hdr_fmt.unpack_file(self._fobj)
print vol_hdr
self.dt = nexrad_to_datetime(vol_hdr.date, vol_hdr.time_ms)
self._version = vol_hdr.version
self.stid = vol_hdr.stid
print self.dt('spot_blanking', 'B'), ('az_index_mode', 'B'),
('num_data_blks', 'H'), ('vol_const_ptr', 'L'),
('el_const_ptr', 'L'), ('rad_const_ptr', 'L'),
('ref_ptr', 'L'), ('vel_ptr', 'L'),
('sw_ptr', 'L'), ('zdr_ptr', 'L'),
('phi_ptr', 'L'), ('rho_ptr', 'L')]
msg31_data_hdr_fmt = NamedStruct(msg31_data_hdr_fields, '>', 'Msg31DataHdr')
msg31_vol_const_fields = [('type', 's'), ('name', '3s'), ('size', 'H'),
('major', 'B'), ('minor', 'B'), ('lat', 'f'),
('lon', 'f'), ('site_amsl', 'h'),
('feedhorn_agl', 'H'), ('calib_dbz', 'f'),
('txpower_h', 'f'), ('txpower_v', 'f'),
('sys_zdr', 'f'), ('phidp0', 'f'),
('vcp', 'H'), (None, '2x')]
msg31_vol_const_fmt = NamedStruct(msg31_vol_const_fields, '>', 'VolConsts')
msg31_el_const_fields = [('type', 's'), ('name', '3s'), ('size', 'H'),
('atmos_atten', 'h'), ('calib_dbz0', 'f')]
msg31_el_const_fmt = NamedStruct(msg31_el_const_fields, '>', 'ElConsts')
rad_const_fields = [('type', 's'), ('name', '3s'), ('size', 'H'),
('unamb_range', 'H'), ('noise_h', 'f'),
('noise_v', 'f'), ('nyq_vel', 'H'), (None, '2x')]
rad_const_fmt = NamedStruct(rad_const_fields, '>', 'RadConsts')
def _decode_msg31(self, msg):
print len(msg)
data_hdr = self.msg31_data_hdr_fmt.unpack_from(msg)
print data_hdr
vol_consts = self.msg31_vol_const_fmt.unpack_from(msg,
data_hdr.vol_const_ptr)
print vol_constsdata_hdr = self.msg31_data_hdr_fmt.unpack_from(msg)
print data_hdr
vol_consts = self.msg31_vol_const_fmt.unpack_from(msg,
data_hdr.vol_const_ptr)
print vol_consts
el_consts = self.msg31_el_const_fmt.unpack_from(msg,
data_hdr.el_const_ptr)
print el_consts
rad_consts = self.rad_const_fmt.unpack_from(msg, data_hdr.rad_const_ptr)
print rad_consts
def _decode_msg1(self, msg):
pass
class Level3File(object):
header_fmt = NamedStruct([('code', 'H'), ('date', 'H'), ('time', 'l'),
('msg_len', 'L'), ('src_id', 'h'), ('dest_id', 'h'),
('num_blks', 'H')], '>', 'MsgHdr')
prod_desc_fmt = NamedStruct([('divider', 'h'), ('lat', 'l'), ('lon', 'l'),
('height', 'h'), ('prod_code', 'h'), ('op_mode', 'h'),
('vcp', 'h'), ('seq_num', 'H'), ('vol_num', 'H'),
('vol_date', 'H'), ('vol_start_time', 'l'), ('prod_gen_date', 'H'),
('prod_gen_time', 'l'), ('dep1', 'h'), ('dep2', 'h'), ('el_num', 'H'),
('dep3', 'h'), ('thr1', 'H'), ('thr2', 'H'), ('thr3', 'H'),
('thr4', 'H'), ('thr5', 'H'), ('thr6', 'H'), ('thr7', 'H'),
('thr8', 'H'), ('thr9', 'H'), ('thr10', 'H'), ('thr11', 'H'),
('thr12', 'H'), ('thr13', 'H'), ('thr14', 'H'), ('thr15', 'H'),
('thr16', 'H'), ('dep4', 'h'), ('dep5', 'h'), ('dep6', 'h'),
('dep7', 'h'), ('dep8', 'h'), ('dep9', 'h'), ('dep10', 'h'),
('n_maps', 'h'), ('sym_off', 'L'), ('graph_off', 'L'),
('tab_off', 'L')], '>', 'ProdDesc')
sym_block_fmt = NamedStruct([('divider', 'h'), ('block_id', 'h'),MetPy
Collection of tools for reading, visualizing and performing calculations with weather data.
Package Health Score
90 / 100Popular MetPy functions
- metpy.calc
- metpy.calc.coriolis_parameter
- metpy.calc.get_wind_components
- metpy.calc.lat_lon_grid_deltas
- metpy.calc.parcel_profile
- metpy.calc.smooth_n_point
- metpy.calc.wind_components
- metpy.cbook.get_test_data
- metpy.io.nexrad.NamedStruct
- metpy.plots.add_metpy_logo
- metpy.plots.SkewT
- metpy.plots.StationPlot
- metpy.testing.assert_almost_equal
- metpy.testing.assert_array_almost_equal
- metpy.testing.assert_array_equal
- metpy.units.masked_array
- metpy.units.units
- metpy.units.units.degC
- metpy.units.units.hPa
- metpy.units.units.knots