Updates for S5P processing. (!32) · Merge requests · CAMS / CSO

config/tutorial/tutorial.rc

+6 −1

Changes for config/tutorial/tutorial.rc: 6 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -308,6 +308,8 @@ cso.tutorial.inquire-table-dataspace.url : https://stac.data
		! collection name:
		cso.tutorial.inquire-table-dataspace.collections : sentinel-5p-l2-no2-rpro \
		sentinel-5p-l2-no2-offl
		!!~ near-real time processing:
		!cso.tutorial.inquire-table-dataspace.collections : sentinel-5p-l2-no2-nrti

		! target area;
		!!~ empty for no limitation:
		@@ -375,6 +377,8 @@ my.tutorial.selection : C03
		!
		cso.tutorial.convert.selection : (%{collection} == '03') and (%{processing} == 'RPRO') ; \
		(%{collection} == '03') and (%{processing} == 'OFFL')
		!!~ for near-real-time processing:
		!cso.tutorial.convert.selection : %{collection} == '03'

		! input directory;
		! files are searched here or downloaded to if not present yet;
		@@ -425,7 +429,8 @@ cso.tutorial.convert.filter.quality.units : 1
		! - times are taken from mid of selection, rounded to hours
		! - use '%{processing}' for the processing name
		! - use '%{orbit}' for orbit number
		cso.tutorial.convert.output.filename : ${my.work}/CSO-data/${my.region}/S5p/NO2/${my.tutorial.selection}/%Y/%m/S5p_%{processing}_NO2_%{orbit}.nc
		! - include start time to have different files if orbit is stored in patches
		cso.tutorial.convert.output.filename : ${my.work}/CSO-data/${my.region}/S5p/NO2/${my.tutorial.selection}/%Y/%m/S5p_%{processing}_NO2_%{orbit}_%Y%m%dT%H%M%S.nc

		! pack variables on output:
		cso.tutorial.convert.output.packed : True

src/cso/cso_colhub.py

+70 −9

Changes for src/cso/cso_colhub.py: 70 added lines, 9 removed lines.
Original line number	Diff line number	Diff line
		@@ -29,6 +29,8 @@
		#
		# 2026-04, Arjo Segers
		# Initialize columns of listing file.
		# Added production time of data files to listings.
		# Select file with latest production time if multiple are found.
		#

		########################################################################
		@@ -185,7 +187,7 @@ class CSO_ColHubMirror_Inquire(utopya.UtopyaRc):

		# initiallize for (re)creation,
		# define extra columns to ensure correct content even for empty files:
		listing = cso_file.CSO_Listing( columns=["orbit","processing","collection","processor_version","href"] )
		listing = cso_file.CSO_Listing( columns=["orbit","processing","collection","processor_version","production_time","href"] )

		# archive directories
		archive_dirs = self.GetSetting("dir").split()
		@@ -242,7 +244,7 @@ class CSO_ColHubMirror_Inquire(utopya.UtopyaRc):
		orbit,
		collection,
		processor_version,
		prod_time,
		production_time,
		) = rest.split("_")
		except:
		logging.error(
		@@ -267,6 +269,7 @@ class CSO_ColHubMirror_Inquire(utopya.UtopyaRc):
		data["processing"] = processing
		data["collection"] = collection
		data["processor_version"] = processor_version
		data["production_time"] = production_time
		data["href"] = os.path.join(root, fname)
		# update record:
		listing.UpdateRecord(fname, data, indent=f"{indent} ")
		@@ -413,8 +416,10 @@ class CSO_ColHubMirror_Missing(utopya.UtopyaRc):
		# info ..
		logging.info(f"{indent}create %s ..." % lst_file)

		# initiallize for (re)creation:
		listing = cso_file.CSO_Listing()
		# initiallize for (re)creation,
		# define extra columns to ensure correct content even for empty files:
		listing = cso_file.CSO_Listing( columns=["orbit","processing","collection","processor_version","production_time","href"] )


		# table with all available files:
		listfile_all = self.GetSetting("all.file")
		@@ -460,6 +465,8 @@ class CSO_ColHubMirror_Missing(utopya.UtopyaRc):
		logging.info(f"{indent}loop over available orbits ...")
		# loop:
		for orbit in orbits:
		## testing ...
		#if orbit != '40182' : continue
		# info ..
		logging.info(f"{indent} orbit '{orbit}' ...")

		@@ -469,11 +476,66 @@ class CSO_ColHubMirror_Missing(utopya.UtopyaRc):
		# nothing selected?
		if len(xlst) == 0:
		continue
		#
		elif len(xlst) == 1:
		# single record:
		rec = xlst.GetRecord(irec=0)
		#
		else:
		# multiple records ...
		# init name of selected file:
		selected_filename = None
		#
		# 1st adhoc fix ...
		# check if these are the same data but file was recreated later (and forgotten to be removed ...)
		# S5P_RPRO_L2__CO_____20250715T142651_20250715T161018_40182_03_020800_20251103T112819.nc
		# S5P_RPRO_L2__CO_____20250715T142651_20250715T161018_40182_03_020800_20251030T133312.nc
		# init basenames:
		basenames = []
		# loop over filenames:
		for filename in xlst.df.index:
		# basename without production time:
		basename = filename.rsplit(sep="_",maxsplit=1)[0]
		# store:
		if basename not in basenames:
		basenames.append( basename )
		#endif
		# first?
		if selected_filename is None:
		# store first values:
		selected_filename = filename
		else:
		# newer?
		if filename > selected_filename:
		selected_filename = filename
		#endif
		#endif
		#endfor # filenames
		# should be the same ..
		if len(basenames) == 1:
		# info ...
		logging.info(f"{indent} selected single file based on latest production time:")
		logging.info(f"{indent} {selected_filename}")
		else:
		# info ...
		logging.warning(f"{indent} checked on production time, but found multiple base names:")
		for basename in basenames:
		logging.warning(f"{indent} {basename}")
		#endfor
		# reset selected name, check below:
		selected_filename = None
		#endif
		#
		# none of the adhoc fix lead to selection of a single file ...
		if selected_filename is None:
		logging.error(f"multiple records selected, something wrong with files or selection?")
		raise Exception
		#endif
		#
		# extract record:
		rec = xlst.GetRecord( name=selected_filename )
		# endif

		# loop over selected records:
		for irec in range(len(xlst)):
		# current:
		rec = xlst.GetRecord(irec)
		## info ..
		# logging.info( f"{indent} {fname}" )
		# set flag:
		@@ -497,7 +559,6 @@ class CSO_ColHubMirror_Missing(utopya.UtopyaRc):
		# add record to list:
		listing.UpdateRecord(rec["filename"], rec)
		# endif
		# endfor

		## TESTING ...
		# if len(listing) > 0 :

src/cso/cso_dataspace.py

+14 −1

Changes for src/cso/cso_dataspace.py: 14 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -55,6 +55,8 @@
		# Fixed definition of bounding box for global selection.
		# Added increasing delays to avoid rate limit errors from STAC catalogue inquiry.
		# Check on different paths to 'href' attribute in stac inquiry.
		# Trap files that could not be downloaded.
		# Added production time of data files to listings.
		#


		@@ -378,7 +380,7 @@ class CSO_DataSpace_Inquire(utopya.UtopyaRc):
		platform_name, processing, rest = product_id.split("_", 2)
		product_type = rest[0:10]
		parts = rest[11:].split("_")
		start_time, end_time, orbit, collection, processor_version, prod_time = parts
		start_time, end_time, orbit, collection, processor_version, production_time = parts

		# convert:
		tfmt = "%Y%m%dT%H%M%S"
		@@ -427,6 +429,7 @@ class CSO_DataSpace_Inquire(utopya.UtopyaRc):
		"processing": [processing],
		"collection": [collection],
		"processor_version": [processor_version],
		"production_time": [production_time],
		"filename": [filename],
		"href": [href],
		}
		@@ -783,8 +786,12 @@ class CSO_DataSpace_Downloader(object):
		# check ...
		if npfile != 1:
		print(f"ERROR - found {npfile} files in S3 bucket for product: {product}")
		if npfile == 0:
		raise Exception(f"file not found in S3 bucket")
		else:
		raise Exception
		#endif
		#endif

		# loop over (single) files:
		for pfile in pfiles:
		@@ -893,8 +900,14 @@ class CSO_DataSpace_Downloader(object):

		# unknown ...
		except Exception as err:
		# check on known errors ..
		if "file not found in S3 bucket" in str(err):
		logging.warning(f"{indent} WARNING - could not download: {href}")
		break
		else:
		logging.error(f"{indent}unknown error:")
		logging.error(f"{indent} {str(err)}")
		#endif

		# endtry

src/cso/cso_file.py

+21 −6

Changes for src/cso/cso_file.py: 21 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -68,6 +68,9 @@
		# 2026-04, Arjo Segers
		# Added optional "columns" argument to initialize empty listing object.
		#
		# 2026-05, Arjo Segers
		# Support `name` argument for GetRecord method.
		#

		########################################################################
		###
		@@ -1407,11 +1410,24 @@ class CSO_Listing(object):

		# *

		def GetRecord(self, irec):
		def GetRecord(self, irec=None, name=None):
		"""
		Return record for corresponding 0-based row index.
		Return record for corresponding 0-based row index or index name.
		"""

		# modules:
		import numpy

		# by name?
		if name is not None:
		indx, = numpy.where( self.df.index == name )
		if len(indx) == 0:
		logging.error(f"value `{name}` not found in index")
		raise Exception
		#endif
		irec = indx[0]
		#endif

		# extract:
		rec = self.df.iloc[irec].copy()
		# add index values:
		@@ -1604,8 +1620,7 @@ class CSO_Listing(object):
		(%{processor_version} == '020400') & (%{processing} == 'RPRO') ; ...

		This is evaluted after the previously described selections.
		The result should be None or exactly one record.
		Eventually skip files in ``blacklists``.
		Eventually skip files in ``blacklist``.

		"""

		@@ -1650,7 +1665,7 @@ class CSO_Listing(object):
		# endif
		# select:
		df = df[df[key] == value]
		# endfor
		# endfor # key/value pairs

		# *

		@@ -1734,7 +1749,7 @@ class CSO_Listing(object):
		df = df.loc[selected]
		# endif

		# endif
		# endif # expr defined

		# *

src/cso/cso_s5p.py

+85 −43

Changes for src/cso/cso_s5p.py: 85 added lines, 43 removed lines.
Original line number	Diff line number	Diff line
		@@ -81,6 +81,13 @@
		# 2026-03, Arjo Segers
		# Updated selection of download source for Copernicus Dataspace.
		#
		# 2026-04, Arjo Segers
		# Trap download errors.
		# Select input file with latest processor version or production time if multiple are selected.
		#
		# 2026-05, Arjo Segers
		# Convert multiple files if orbit is stored in patches. Check on duplicate output files.
		#

		########################################################################
		###
		@@ -2865,7 +2872,7 @@ class CSO_S5p_Convert(utopya.UtopyaRc):
		# loop over orbits:
		for orbit in orbits:
		# info ...
		logging.info(indent + ' orbit "%s" ...' % orbit)
		logging.info(f"{indent} orbit '{orbit}' ...")

		## TESTING ..
		#if orbit != "21861":
		@@ -2879,14 +2886,18 @@ class CSO_S5p_Convert(utopya.UtopyaRc):
		)
		# no orbit found? next:
		if len(odf) == 0:
		# next orbit:
		continue
		elif len(odf) > 1:
		logging.error(f"found multiple records matching selection;" +
		" use finer selection, or something wrong in inquiry table(s)?" )
		raise Exception
		#endif

		# keep list of target files to check on duplicates ..
		output_filenames = []
		# loop over records:
		for irec in range(len(odf)):
		# info ...
		logging.info(f"{indent} record {irec+1}/{len(odf)} ...")
		# selected record:
		rec = odf.GetRecord(0)
		rec = odf.GetRecord(irec)

		# start time of orbit:
		t0 = rec["start_time"]
		@@ -2895,16 +2906,28 @@ class CSO_S5p_Convert(utopya.UtopyaRc):
		output_filename = t0.strftime(output_filename__template)
		for key in rec.keys():
		if type(rec[key]) == str:
		output_filename = output_filename.replace("%{" + key + "}", rec[key])
		output_filename = output_filename.replace(f"%{{{key}}}", rec[key])
		# endif
		# endfor
		# info ...
		logging.info(" target file: %s" % output_filename)
		logging.info(f"{indent} target file: {output_filename}")

		# check ..
		if output_filename in output_filenames:
		logging.error(f"output file already created for this record:")
		logging.error(f" {output_filename}")
		logging.error(f"Orbit stored in more than one file?")
		logging.error(f"Then ensure to include a start-time in the output filename template:")
		logging.error(f" {output_filename__template}")
		raise Exception
		#endif
		# store:
		output_filenames.append( output_filename )

		# selected productions?
		if len(whitelist) > 0:
		if os.path.basename(output_filename) not in whitelist:
		logging.info(" not in whitelist, skip ...")
		logging.info("{indent} not in whitelist, skip ...")
		continue
		# endif
		# endif
		@@ -2922,7 +2945,7 @@ class CSO_S5p_Convert(utopya.UtopyaRc):
		# error file present?
		if with_error_files and os.path.isfile(output_errfile):
		# info ..
		logging.info(" error file present:")
		logging.info(f"{indent} error file present:")
		# read:
		with open(output_errfile, "r") as f:
		lines = f.readlines()
		@@ -2932,38 +2955,38 @@ class CSO_S5p_Convert(utopya.UtopyaRc):
		logging.info(" %s" % line.strip())
		# endfor
		# info ..
		logging.info(" do not try to create again ...:")
		logging.info(f"{indent} do not try to create again ...:")
		# do not create ..
		create = False
		#
		# ~ message file present?
		elif os.path.isfile(output_msgfile):
		# info ..
		logging.info(" message file present:")
		logging.info(f"{indent} message file present:")
		# read:
		with open(output_msgfile, "r") as f:
		lines = f.readlines()
		# endwith
		# info..
		for line in lines:
		logging.info(" %s" % line.strip())
		logging.info(f"{indent} {line.strip()}")
		# endfor
		# info ..
		logging.info(" do not try to create again ...:")
		logging.info(f"{indent} do not try to create again ...:")
		# do not create ..
		create = False
		#
		# ~ not present yet, create file?
		elif not os.path.isfile(output_filename):
		# info ...
		logging.info(" create new file ...")
		logging.info(f"{indent} create new file ...")
		# always create:
		create = True
		#
		# ~ renew?
		elif renew:
		# info ...
		logging.info(" renew file ...")
		logging.info(f"{indent} renew file ...")
		# always create:
		create = True
		#
		@@ -2977,7 +3000,7 @@ class CSO_S5p_Convert(utopya.UtopyaRc):
		orbit_file = csf.GetAttr("orbit_file", quiet=True)
		if orbit_file is None:
		# info ...
		logging.info(' replace existing file, no "orbit_file" attribute yet ...')
		logging.info(f"{indent} replace existing file, no 'orbit_file' attribute yet ...")
		# attribute not found, re-create to have this attribute:
		create = True
		elif fname <= orbit_file:
		@@ -2987,9 +3010,7 @@ class CSO_S5p_Convert(utopya.UtopyaRc):
		create = False
		else:
		# info ...
		logging.info(
		" replace existing file, based on older processing: %s" % orbit_file
		)
		logging.info(f"{indent} replace existing file, based on older processing: {orbit_file}")
		# replace:
		create = True
		# endif
		@@ -3007,7 +3028,7 @@ class CSO_S5p_Convert(utopya.UtopyaRc):
		# this is the input file:
		input_file = href
		# info ..
		logging.info(" input file: %s" % input_file)
		logging.info(f"{indent} input file: {input_file}")

		else:
		# input dir:
		@@ -3015,18 +3036,18 @@ class CSO_S5p_Convert(utopya.UtopyaRc):
		# replace templates:
		for key in rec.keys():
		if type(rec[key]) == str:
		input_dir = input_dir.replace("%{" + key + "}", rec[key])
		input_dir = input_dir.replace(f"%{{{key}}}", rec[key])
		# endif
		# endfor
		# full path:
		input_file = os.path.join(input_dir, rec["filename"])

		# info ..
		logging.info(" input file: %s" % input_file)
		logging.info(f"{indent} input file: {input_file}")
		# check ..
		if not os.path.isfile(input_file):
		# info ..
		logging.info(" not present yet, download ...")
		logging.info(f"{indent} not present yet, download ...")
		# download url:
		href = rec["href"]
		# initialize download?
		@@ -3162,11 +3183,13 @@ class CSO_S5p_Convert(utopya.UtopyaRc):

		# endif # create

		# endif # irec (patches in orbit)

		## testing ...
		# logging.warning( 'break after first orbit' )
		# break

		# endfor # input files
		# endfor # orbits

		# info ...
		logging.info(f"{indent}")
		@@ -3326,7 +3349,7 @@ class CSO_S5p_Download(utopya.UtopyaRc):
		df.sort_values("filename", inplace=True)

		# info ...
		logging.info(f"{indent}number of files : %i" % len(df))
		logging.info(f"{indent}number of files : {len(df)}")

		# list of ';' seperated selection expression:
		# (%{processor_version} == '020400') & (%{processing} == 'RPRO') ; ...
		@@ -3334,14 +3357,14 @@ class CSO_S5p_Download(utopya.UtopyaRc):
		# replace templates:
		# (xrec['processor_version'] == '020400') & (xrec['processing'] == 'RPRO') ; ...
		for key in df.keys():
		line = line.replace("%{" + key + "}", "xrec['" + key + "']")
		line = line.replace(f"%{{{key}}}", f"xrec['{key}']")
		# endfor
		# split:
		selections = line.split(";")
		# info ..
		logging.info("selection criteria (first with matching orbit is used):")
		for selection in selections:
		logging.info(" %s" % selection.strip())
		logging.info(f" {selection.strip()}")
		# endif

		# skip some?
		@@ -3356,7 +3379,9 @@ class CSO_S5p_Download(utopya.UtopyaRc):
		logging.info(f"{indent}found %i orbits with overlap of time range .." % len(xdf))

		# orbit labels:
		orbits = xdf["orbit"].unique()
		orbits = list( xdf["orbit"].unique() )
		# sorted version:
		orbits.sort()

		# no download initialized yet:
		downloader = None
		@@ -3366,11 +3391,14 @@ class CSO_S5p_Download(utopya.UtopyaRc):
		# loop over orbits:
		for orbit in orbits:
		# info ...
		logging.info(indent + ' orbit "%s" ...' % orbit)
		logging.info(f"{indent} orbit '{orbit}'")

		# search for other records for same orbit:
		odf = xdf[xdf["orbit"] == orbit]

		# sort on processor version and production time, newest first:
		odf.sort_values(by=["processor_version","production_time"],ascending=False)

		# storage for status label: "selected", "blacklisted", ...
		filestatus = {}
		# no match yet ..
		@@ -3380,6 +3408,9 @@ class CSO_S5p_Download(utopya.UtopyaRc):
		for selection in selections:
		# make empty again:
		selected = []
		# latest processor version and production time:
		pversion = None
		ptime = None
		# loop over records:
		for indx, xrec in odf.iterrows():
		# skip?
		@@ -3389,20 +3420,32 @@ class CSO_S5p_Download(utopya.UtopyaRc):
		# endif
		# evaluate expression including 'xrec[key]' values:
		if eval(selection):
		# already selected a record? then check on processor version and production time:
		if len(selected) > 0:
		if xrec["processor_version"] < pversion:
		filestatus[xrec["filename"]] = "older processor version"
		continue
		#endif
		if xrec["production_time"] < ptime:
		filestatus[xrec["filename"]] = "older production time"
		continue
		#endif
		#endif
		# store:
		selected.append(xrec["filename"])
		filestatus[xrec["filename"]] = "selected"
		rec = xrec
		pversion = xrec["processor_version"]
		ptime = xrec["production_time"]
		# endif
		# endfor # records
		# exactly one? then leave:
		if len(selected) == 1:
		break
		elif len(selected) > 1:
		logging.error(
		"found more than one orbit file matching selection: %s" % selection
		)
		logging.warning(f"{indent} found more than one orbit file matching selection: {selection}")
		for fname in selected:
		logging.error(" %s" % fname)
		logging.warning(f"{indent} {fname}")
		# endfor
		raise Exception
		# endif # number found
		@@ -3460,7 +3503,7 @@ class CSO_S5p_Download(utopya.UtopyaRc):
		# initialize download?
		if downloader is None:
		# init downloader based on url:
		if "dataspace.copernicus.eu" in href:
		if href.startswith("s3://eodata/"):
		# download from Copernicus DataSpace:
		downloader = cso_dataspace.CSO_DataSpace_Downloader()
		#
		@@ -3469,7 +3512,7 @@ class CSO_S5p_Download(utopya.UtopyaRc):
		downloader = cso_pal.CSO_PAL_Downloader()
		#
		else:
		logging.error("no downloader class defined for url: {href}")
		logging.error(f"no downloader class defined for url: {href}")
		raise Exception
		# endif
		# endif
		@@ -3481,8 +3524,7 @@ class CSO_S5p_Download(utopya.UtopyaRc):

		# download might have failed ..
		if not os.path.isfile(input_file):
		logging.error(f"missing input file")
		logging.error(f" {input_file}")
		logging.error(f"missing input file: {input_file}")
		raise Exception
		# endif

		@@ -3925,7 +3967,7 @@ class CSO_S5p_Download_Listing(utopya.UtopyaRc):
		orbit,
		collection,
		processor_version,
		prod_time,
		production_time,
		) = rest[13:].split("_")
		else:
		product_id = rest[0:10]
		@@ -3935,7 +3977,7 @@ class CSO_S5p_Download_Listing(utopya.UtopyaRc):
		orbit,
		collection,
		processor_version,
		prod_time,
		production_time,
		) = rest[11:].split("_")
		# endif
		except:
		@@ -3958,13 +4000,13 @@ class CSO_S5p_Download_Listing(utopya.UtopyaRc):
		data["orbit"] = orbit
		data["collection"] = collection
		data["processor_version"] = processor_version
		if len(prod_time) == 8:
		if len(production_time) == 8:
		data["processing_time"] = datetime.datetime.strptime(
		prod_time, "%Y%m%d"
		production_time, "%Y%m%d"
		)
		else:
		data["processing_time"] = datetime.datetime.strptime(
		prod_time, tfmt
		production_time, tfmt
		)
		# endif
		# update record:

Updates for S5P processing.

Merge request reports