Skip to content

Instantly share code, notes, and snippets.

@spirrobe

spirrobe/cloudnet_add_serial_number_2_cloudnetfiles.py

Last active September 15, 2023 16:15
Show Gist options
  • Save spirrobe/dde782662bda45feeaeacd15526f062f to your computer and use it in GitHub Desktop.
Save spirrobe/dde782662bda45feeaeacd15526f062f to your computer and use it in GitHub Desktop.
Download ZIP
This is a small scripts that uses netCDF4 to add serial numbers (knownsnos) and instrumenttype (instrumenttypes) to a cloudnet file. It serves as a post-processing tool while cloudnetpy does not yet have that functionality implemented. As such, the instrumenttypes are matched based the filename and this for sure needs to be adjusted for your spe…
Raw
cloudnet_add_serial_number_2_cloudnetfiles.py
#!/usr/bin/env python3
import os
import time
import shutil
import netCDF4 as nc
# get pathwalker from https://gist.github.com/spirrobe/46c3a9d2bbda2475c155065069244969
# and put it somewhere useful, then change the below import to reflect this
# pathwalker in general is case insensitive but this can be switched to case sensitive
# if you need it for other things
from util.pathwalker import pathwalker
path = 'ENTER_PATH_OF_FILES'
# any nc in the folder is (hopefully) a cloudnet file, otherwise refine the pattern/exclude
files = pathwalker(path, extension='nc')
# for single field variables the attr
# source_serial_number
# will be added
# for global attrs the attr
# source_serial_numbers
# will be added
# whether to first copy the file to make a new version in case you want to test
inplace = False
# based on the sources in the field, these serialnumbers are added
# EXTEND THIS BASED ON YOUR USECASE FOR OTHER MODELS / CEILOMETERS / RADARS
knownsnos = {"RPG-Radiometer Physics RPG-FMCW-94": 'RPGFMCWRADARSERIALNUMBER',
"Lufft CHM15k": 'CEILOMETERSERIALNUMBER',
"ECMWF Integrated Forecast System (IFS)": '',
'METEK MIRA-35': '', # as there are several radars, the serialnumber lookup is via instrumenttypes and filename
"RPG-Radiometer Physics HATPRO": 'ENTERRPGHATPROSERIALNUMER', #
}
# a fallback as the knownsnos are meant to llok
instrumenttypes = {"FILEPATTERN_RADARRPGFMCW": 'RPGFMCWRADARSERIALNUMBER',
'FILEPATTERN_RADARMIRA1': 'RADARMIRA1',
'FILEPATTERN_RADARMIRA2': 'RADARMIRA2', # can oftenbe found in the be found in netCDF and for certain types will be included it the first level cloudnet files
'FILEPATTERN_CEILOMETER': 'CEILOMETER_SERIALNUMBER', # can often be found in netCDF and for certain types will be included it the first level cloudnet files
'FILEPATTERN_MWR': 'ENTERRPGHATPROSERIALNUMER',
}
for file in files:
if inplace is False:
shutil.copy(file, file.replace('.nc','_serialnumbersadded.nc'))
# wait until the copy is finished, should be fast enough
time.sleep(0.5)
instrument = file.split(os.sep)[-1].split('_')[0]
# print(instrument)
# ds['target_classification'].setncattr("source", 'test')
ds = nc.Dataset(file, 'r+')
skipkeys = ["latitude", "longitude", "altitude", "time", "height"]
for key in ds.variables.keys():
if key in skipkeys:
continue
if 'source' in ds[key].ncattrs():
sources = ds[key].getncattr('source').split('\n')
snos = []
for s in sources:
if s == 'METEK MIRA-35':
sno = instrument.upper()
else:
sno = knownsnos[s]
snos.append(sno)
snos = '\n'.join(snos)
# print(file, snos.replace('\n',','))
ds[key].setncattr('source_serial_number', snos)
else:
# no source in single field
# check what file type this is
ftype = ds.getncattr("cloudnet_file_type")
if ftype in ['classification', 'categorize',
'drizzle', 'iwc', 'lwc', 'ier', 'der']:
pass
if ftype in ['classification']:
sources = ds.getncattr('source').split('\n')
snos = []
for s in sources:
if s == 'METEK MIRA-35':
sno = instrumenttypes[instrument]
else:
sno = knownsnos[s]
snos.append(sno)
print(f'Adding serial number to {file} for {key}',
f' as {snos} with {sources}\n')
snos = '\n'.join(snos)
ds[key].setncattr('source', '\n'.join(sources))
ds[key].setncattr('source_serial_number', snos)
elif ftype in ['drizzle']:
# radar and lidar
srckey = [0, 1]
sources = ds.getncattr('source').split('\n')
sources = [i[1] for i in enumerate(sources)
if i[0] in srckey]
snos = []
for s in sources:
if s == 'METEK MIRA-35':
sno = instrumenttypes[instrument]
else:
sno = knownsnos[s]
snos.append(sno)
print(f'Adding serial number to {file} for {key}',
f' as {snos} with {sources}\n')
snos = '\n'.join(snos)
sources = '\n'.join(sources)
ds[key].setncattr('source', sources)
ds[key].setncattr('source_serial_number', snos)
# print(sources, snos)
elif ftype in ['iwc', 'ier', 'der']:
# radar and model
srckey = [0, 3]
sources = ds.getncattr('source').split('\n')
sources = [i[1] for i in enumerate(sources)
if i[0] in srckey]
snos = []
for s in sources:
if s == 'METEK MIRA-35':
sno = instrumenttypes[instrument]
else:
sno = knownsnos[s]
snos.append(sno)
print(f'Adding serial number to {file} for {key}',
f' as {snos} with {sources}\n')
snos = '\n'.join(snos)
sources = '\n'.join(sources)
ds[key].setncattr('source', sources)
ds[key].setncattr('source_serial_number', snos)
# print(sources, snos)
elif ftype in ['lwc']:
# mwr and to get clouds the radar
srckey = [0, 2]
sources = ds.getncattr('source').split('\n')
sources = [i[1] for i in enumerate(sources)
if i[0] in srckey]
snos = []
for s in sources:
if s == 'METEK MIRA-35':
sno = instrumenttypes[instrument]
else:
sno = knownsnos[s]
snos.append(sno)
print(f'Adding serial number to {file} for {key}',
f' as {snos} with {sources}\n')
snos = '\n'.join(snos)
sources = '\n'.join(sources)
ds[key].setncattr('source', sources)
ds[key].setncattr('source_serial_number', snos)
pass
else:
snos = []
srckey = -1
if 'beta' in key:
# ceilometer
snos = knownsnos["Lufft CHM15k"]
srckey = [1]
elif 'model' in key:
# model
snos = knownsnos["ECMWF Integrated Forecast System (IFS)"]
srckey = [3]
elif 'radar' in key or 'Z_' in key:
snos = instrumenttypes[instrument]
srckey = [0]
elif 'lwp' in key:
# mwr/hatpro
snos = knownsnos["RPG-Radiometer Physics HATPRO"]
srckey = [2]
elif key in ['category_bits', "insect_prob", "quality_bits"]:
sources = ds.getncattr('source').split('\n')
snos = []
srckey = [0, 1, 2, 3]
for s in sources:
if s == 'METEK MIRA-35':
sno = instrumenttypes[instrument]
else:
sno = knownsnos[s]
snos.append(sno)
else:
print(f'Not adding any serial number to {file} for {key}')
if snos != []:
sources = ds.getncattr('source').split('\n')
sources = [i[1] for i in enumerate(sources)
if i[0] in srckey]
print(f'Adding serial number to {file} for {key}',
f' as {snos} with {sources}\n')
if isinstance(snos, list):
snos = '\n'.join(snos)
ds[key].setncattr('source_serial_number', snos)
if 'source' in ds[key].ncattrs():
pass
else:
ds[key].setncattr('source', '\n'.join(sources))
else:
# print(file, ftype, instrument, key, '****')
# print(instrumenttypes[instrument])
ds[key].setncattr('source_serial_number', instrumenttypes[instrument])
if 'source' in ds.ncattrs():
sources = ds.getncattr('source').split('\n')
snos = []
for s in sources:
if s == 'METEK MIRA-35':
sno = instrument.upper()
else:
sno = knownsnos[s]
snos.append(sno)
snos = '\n'.join(snos)
ds.setncattr('source_serial_numbers', snos)
# if you want to try this out on one file, either pass in just 1 file or uncomment the break below
ds.close()
# break
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment