Skip to content
GitLab
Projects
Groups
Snippets
/
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
geomultisens
gms_preprocessing
Commits
de536db0
Commit
de536db0
authored
Oct 24, 2017
by
Daniel Scheffler
Browse files
Merge branch 'master' into feature/spectral_homogenization
Former-commit-id:
ab157d0d
parents
e5fb4ce3
9767dce1
Changes
22
Hide whitespace changes
Inline
Side-by-side
bin/run_at_geoms__deployed.sh
View file @
de536db0
...
...
@@ -15,13 +15,6 @@ export LD_LIBRARY_PATH=${LD_PATH_PYTHON_GFZ}:${LD_LIBRARY_PATH}
export
PYTHONPATH
=
${
PFX
}
/opt/lib/python3.6/site-packages:
${
PFX
}
/opt/lib/python2.7/site-packages
# Python version must be updated here!
export
GDAL_DATA
=
${
PFX
}
/opt/share/gdal
export
PYTHONPATH
=
${
PYTHONPATH
}
:/home/gfz-fe/scheffler/python_deployed/gms_preprocessing
# needed to find gms_preprocessing
export
PYTHONPATH
=
${
PYTHONPATH
}
:/home/gfz-fe/scheffler/python_deployed/geoarray
# needed to find geoarray
export
PYTHONPATH
=
${
PYTHONPATH
}
:/home/gfz-fe/scheffler/python_deployed/py_tools_ds
# needed to find py_tools_ds
export
PYTHONPATH
=
${
PYTHONPATH
}
:/home/gfz-fe/scheffler/python_deployed/arosics
# needed to find e.g. AROSICS
export
PYTHONPATH
=
${
PYTHONPATH
}
:/home/gfz-fe/hollstein/python/sicor
# needed to find sicor
# execute python script
#ipython --colors='NoColor' run_gms.py jobid "$@" # NoColor must be active for ipython because gwt-log cannot interpret ANSI console colors
python run_gms.py jobid
"
$@
"
...
...
bin/run_gms.py
View file @
de536db0
...
...
@@ -19,7 +19,7 @@ def run_from_jobid(args):
# TODO download: run only the downloader
# set up process controller instance
PC
=
process_controller
(
args
.
jobid
,
parallelization_level
=
'scenes'
)
PC
=
process_controller
(
args
.
jobid
,
parallelization_level
=
'scenes'
,
db_host
=
'geoms'
)
# FIXME hardcoded host
# PC.job.path_procdata_scenes = '/geoms/data/processed_scenes_dev'
# PC.job.path_procdata_MGRS = '/geoms/data/processed_mgrs_tiles_dev'
...
...
bin/run_gms.sh
0 → 100644
View file @
de536db0
#!/usr/bin/env bash
# execute python script
# NOTE: This asserts that the gms_preprocessing has been installed via 'pip install...' and that the PATH environment
# variable points to the correct Python interpreter.
#ipython --colors='NoColor' run_gms.py jobid "$@" # NoColor must be active for ipython because gwt-log cannot interpret ANSI console colors
run_gms.py jobid
"
$@
"
gms_preprocessing/__init__.py
View file @
de536db0
...
...
@@ -13,8 +13,8 @@ from .processing.process_controller import process_controller # noqa: E402
__author__
=
"""Daniel Scheffler"""
__email__
=
'daniel.scheffler@gfz-potsdam.de'
__version__
=
'0.
8.5
'
__versionalias__
=
'201710
10
.0
1
'
__version__
=
'0.
9.1
'
__versionalias__
=
'201710
23
.0
4
'
__all__
=
[
'algorithms'
,
'io'
,
'misc'
,
...
...
gms_preprocessing/algorithms/L1A_P.py
View file @
de536db0
...
...
@@ -34,7 +34,7 @@ class L1A_object(GMS_object):
"""Features input reader and raster-/metadata homogenization."""
def
__init__
(
self
,
image_type
=
''
,
satellite
=
''
,
sensor
=
''
,
subsystem
=
''
,
sensormode
=
''
,
acq_datetime
=
None
,
entity_ID
=
''
,
scene_ID
=-
9999
,
filename
=
''
,
dataset_ID
=-
9999
):
entity_ID
=
''
,
scene_ID
=-
9999
,
filename
=
''
,
dataset_ID
=-
9999
,
**
kwargs
):
""":param : instance of gms_object.GMS_object or None
"""
# TODO docstring
...
...
@@ -203,6 +203,7 @@ class L1A_object(GMS_object):
i
,
list_matching_dsIdx
=
0
,
[]
while
True
:
# Get dataset indices within HDF file
# noinspection PyBroadException
try
:
ds
=
hdfFile
.
select
(
i
)
if
subsystem_identifier
in
str
(
ds
.
dimensions
())
and
'ImagePixel'
in
str
(
ds
.
dimensions
()):
...
...
@@ -496,10 +497,11 @@ class L1A_object(GMS_object):
'mask_nodata'
)
and
self
.
mask_nodata
is
not
None
,
"The L1A object needs to have a nodata mask."
self
.
logger
.
info
(
'Calculating true data corner positions (image and world coordinates)...'
)
if
re
.
search
(
'ETM+'
,
self
.
sensor
)
and
self
.
acq_datetime
>
datetime
.
datetime
(
year
=
2003
,
month
=
5
,
day
=
31
,
tzinfo
=
datetime
.
timezone
.
utc
):
# if re.search('ETM+', self.sensor) and self.acq_datetime > datetime.datetime(year=2003, month=5, day=31,
# tzinfo=datetime.timezone.utc):
if
is_dataset_provided_as_fullScene
(
self
.
GMS_identifier
):
self
.
trueDataCornerPos
=
calc_FullDataset_corner_positions
(
self
.
mask_nodata
,
algorithm
=
'numpy'
,
assert_four_corners
=
Fals
e
)
assert_four_corners
=
Tru
e
)
else
:
self
.
trueDataCornerPos
=
calc_FullDataset_corner_positions
(
self
.
mask_nodata
,
assert_four_corners
=
False
)
...
...
@@ -546,9 +548,10 @@ class L1A_object(GMS_object):
def
calc_orbit_overpassParams
(
self
):
"""Calculate orbit parameters."""
self
.
MetaObj
.
overpassDurationSec
,
self
.
MetaObj
.
scene_length
=
\
self
.
MetaObj
.
get_overpassDuration_SceneLength
(
self
.
fullSceneCornerLonLat
,
self
.
fullSceneCornerPos
,
self
.
shape_fullArr
,
self
.
logger
)
if
is_dataset_provided_as_fullScene
(
self
.
GMS_identifier
):
self
.
MetaObj
.
overpassDurationSec
,
self
.
MetaObj
.
scene_length
=
\
self
.
MetaObj
.
get_overpassDuration_SceneLength
(
self
.
fullSceneCornerLonLat
,
self
.
fullSceneCornerPos
,
self
.
shape_fullArr
,
self
.
logger
)
def
add_rasterInfo_to_MetaObj
(
self
,
custom_rasObj
=
None
):
"""
...
...
@@ -602,13 +605,15 @@ class L1A_object(GMS_object):
def
calc_mean_VAA
(
self
):
"""Calculates mean viewing azimuth angle using sensor flight line derived from full scene corner coordinates."""
if
re
.
search
(
'Sentinel-2'
,
self
.
satellite
,
re
.
I
):
if
is_dataset_provided_as_fullScene
(
self
.
GMS_identifier
):
self
.
VAA_mean
=
\
GEOP
.
calc_VAA_using_fullSceneCornerLonLat
(
self
.
fullSceneCornerLonLat
,
self
.
MetaObj
.
orbitParams
)
else
:
# e.g. Sentinel-2 / RapidEye
self
.
logger
.
warning
(
'No precise calculation of mean viewing azimuth angle possible because orbit track '
'cannot be reconstructed from dataset since full scene corner positions are unknown. '
'Mean VAA angle is filled with the mean value of the viewing azimuth array provided '
'in metadata.'
)
self
.
VAA_mean
=
self
.
MetaObj
.
IncidenceAngle
else
:
self
.
VAA_mean
=
\
GEOP
.
calc_VAA_using_fullSceneCornerLonLat
(
self
.
fullSceneCornerLonLat
,
self
.
MetaObj
.
orbitParams
)
self
.
logger
.
info
(
'Calculation of mean VAA...: %s'
%
round
(
self
.
VAA_mean
,
2
))
gms_preprocessing/algorithms/L1C_P.py
View file @
de536db0
...
...
@@ -259,7 +259,7 @@ class AtmCorr(object):
path_logfile
=
inObj
.
pathGen
.
get_path_logfile
()
fileHandler
=
logging
.
FileHandler
(
path_logfile
,
mode
=
'a'
)
fileHandler
.
setFormatter
(
logger_atmCorr
.
formatter_fileH
)
fileHandler
.
setLevel
(
logging
.
DEBUG
)
fileHandler
.
setLevel
(
CFG
.
job
.
log_level
)
logger_atmCorr
.
addHandler
(
fileHandler
)
...
...
gms_preprocessing/algorithms/geoprocessing.py
View file @
de536db0
...
...
@@ -1163,9 +1163,12 @@ def calc_VAA_using_fullSceneCornerLonLat(fullSceneCornerLonLat, orbit_params):
"""Calculates the Viewing azimuth angle (defined as 90 degrees from the flight line),
e.g. if flight line is 8 degrees from North -> VAA will be 98 degrees.
:param fullSceneCornerLonLat:
:param fullSceneCornerLonLat:
UL, UR, LL, LR
:param orbit_params: list of [altitude, inclination, period] => inclination is used as fallback
"""
assert
len
(
fullSceneCornerLonLat
)
==
4
,
\
'VAA can only be calculated with fullSceneCornerLonLat representing 4 coordinates (UL, UR, LL, LR).'
UL_LonLat
,
UR_LonLat
,
LL_LonLat
,
LR_LonLat
=
fullSceneCornerLonLat
forward_az_left
=
pyproj
.
Geod
(
ellps
=
'WGS84'
).
inv
(
*
LL_LonLat
,
*
UL_LonLat
)[
0
]
forward_az_right
=
pyproj
.
Geod
(
ellps
=
'WGS84'
).
inv
(
*
LR_LonLat
,
*
UR_LonLat
)[
0
]
...
...
@@ -1175,7 +1178,7 @@ def calc_VAA_using_fullSceneCornerLonLat(fullSceneCornerLonLat, orbit_params):
if
abs
(
VAA_mean
-
90
)
<
1
:
# fullSceneCornerLonLat obviously don't belong to a full scene but a granule
assert
orbit_params
warnings
.
warn
(
'Derivation of mean VAA angle from flight line delivered a non reasonable value (%s degre
s
s).'
warnings
.
warn
(
'Derivation of mean VAA angle from flight line delivered a non reasonable value (%s degre
e
s).'
'Using sensor inclination (%s degrees) as fallback.'
%
(
VAA_mean
,
orbit_params
[
1
]))
VAA_mean
=
float
(
orbit_params
[
1
])
# inclination # FIXME is this correct?
...
...
@@ -1304,7 +1307,7 @@ def calc_SZA_SAA_array(shape_fullArr, arr_pos, AcqDate, CenterAcqTime, fullScene
:param AcqDate:
:param CenterAcqTime:
:param fullSceneCornerPos:
:param fullSceneCornerLonLat:
:param fullSceneCornerLonLat:
UL, UR, LL, LR
:param overpassDurationSec:
:param logger:
:param meshwidth: <int> defines the density of the mesh used for generating the output
...
...
gms_preprocessing/config.py
View file @
de536db0
...
...
@@ -187,7 +187,7 @@ class Job(object):
self
.
path_job_logs
=
self
.
DB_config
[
'path_job_logs'
]
else
:
# in test mode, the repository should be self-contained -> use only relative paths
self
.
path_
archi
ve
=
self
.
absP
(
'../tests/data/'
)
self
.
path_
fileser
ve
r
=
self
.
absP
(
'../tests/data/'
)
self
.
path_archive
=
self
.
absP
(
'../tests/data/archive_data/'
)
self
.
path_procdata_scenes
=
self
.
absP
(
'../tests/data/output_scenes/'
)
self
.
path_procdata_MGRS
=
self
.
absP
(
'../tests/data/output_mgrs_tiles/'
)
...
...
@@ -324,7 +324,8 @@ class Usecase:
self
.
scale_factor_TOARef
=
10000
self
.
scale_factor_BOARef
=
10000
self
.
scale_factor_errors_ac
=
255
self
.
virtual_sensor_id
=
10
# Sentinel-2A 10m
# self.virtual_sensor_id = 10 # Sentinel-2A 10m
self
.
virtual_sensor_id
=
1
# Landsat-8
self
.
datasetid_spectral_ref
=
249
# Sentinel-2A
self
.
target_CWL
=
[]
self
.
target_FWHM
=
[]
...
...
@@ -339,16 +340,17 @@ class Usecase:
self
.
sort_bands_by_cwl
=
int
(
query_cfg
(
'sort_bands_by_cwl'
))
self
.
conversion_type_optical
=
query_cfg
(
'conversion_type_optical'
)
self
.
conversion_type_thermal
=
query_cfg
(
'conversion_type_thermal'
)
self
.
datasetid_spatial_ref
=
query_job
(
'datasetid_spatial_ref'
)
self
.
virtual_sensor_id
=
query_job
(
'virtualsensorid'
)
self
.
virtual_sensor_id
=
self
.
virtual_sensor_id
if
self
.
virtual_sensor_id
!=
-
1
else
10
# Sentinel-2A 10m
self
.
datasetid_spatial_ref
=
query_job
(
'datasetid_spatial_ref'
)
# self.datasetid_spatial_ref = 104
self
.
virtual_sensor_name
=
query_vir
(
'name'
,
self
.
virtual_sensor_id
)
self
.
datasetid_spectral_ref
=
query_vir
(
'spectral_characteristics_datasetid'
,
self
.
virtual_sensor_id
)
# FIXME column is empty a known datasetid as spectral characteristics virtual sensor is chosen:
self
.
target_CWL
=
query_vir
(
'wavelengths_pos'
,
self
.
virtual_sensor_id
)
# FIXME column is empty a known datasetid as spectral characteristics virtual sensor is chosen:
self
.
target_FWHM
=
query_vir
(
'band_width'
,
self
.
virtual_sensor_id
)
# FIXME target GSD setting is a duplicate to datasetid_spatial_ref!
self
.
target_gsd
=
query_vir
(
'spatial_resolution'
,
self
.
virtual_sensor_id
)
# table features only 1 value for X/Y-dims
self
.
target_gsd
=
xgsd
,
ygsd
=
\
...
...
@@ -497,7 +499,7 @@ class Usecase:
scenes.acquisitiondate,
scenes.entityid,
scenes.filename,
COALESCE(scenes_proc.proc_level::text, 'L
0
A') AS proc_level,
COALESCE(scenes_proc.proc_level::text, 'L
1
A') AS proc_level,
datasets.image_type,
satellites.name AS satellite,
sensors.name AS sensor,
...
...
@@ -517,7 +519,7 @@ class Usecase:
ds
=
OrderedDict
()
ds
[
"proc_level"
]
=
row
[
"proc_level"
]
ds
[
"scene_ID"
]
=
row
[
"sceneid"
]
ds
[
"dataset
id
"
]
=
row
[
"datasetid"
]
ds
[
"dataset
_ID
"
]
=
row
[
"datasetid"
]
ds
[
"image_type"
]
=
row
[
"image_type"
]
ds
[
"satellite"
]
=
row
[
"satellite"
]
ds
[
"sensor"
]
=
row
[
"sensor"
]
...
...
gms_preprocessing/misc/database_tools.py
View file @
de536db0
...
...
@@ -1136,8 +1136,8 @@ def delete_processing_results(scene_ID, proc_level='all', force=False):
try
:
shutil
.
rmtree
(
path_procdata
)
except
OSError
:
# directory not deletable because it is not empty
if
[
F
for
F
in
glob
.
glob
(
path_procdata
)
if
not
F
.
startswith
(
'.fuse_hidden'
)]:
raise
# raise OSError if there are other files tha
t
.fuse_hidden... remaining
if
[
F
for
F
in
glob
.
glob
(
path_procdata
)
if
not
os
.
path
.
basename
(
F
)
.
startswith
(
'.fuse_hidden'
)]:
raise
# raise OSError if there are other files tha
n
.fuse_hidden... remaining
else
:
files2delete
=
glob
.
glob
(
os
.
path
.
join
(
path_procdata
,
'*%s*'
%
proc_level
))
errors
=
False
# default
...
...
gms_preprocessing/misc/definition_dicts.py
View file @
de536db0
...
...
@@ -153,7 +153,7 @@ def is_dataset_provided_as_fullScene(GMS_identifier):
sensorcode
=
get_GMS_sensorcode
(
GMS_identifier
)
dict_fullScene_or_tiles
=
{
'AVNIR-2'
:
True
,
'AST_full'
:
Fals
e
,
'AST_full'
:
Tru
e
,
'AST_V1'
:
True
,
'AST_V2'
:
True
,
'AST_S'
:
True
,
...
...
@@ -173,7 +173,7 @@ def is_dataset_provided_as_fullScene(GMS_identifier):
'SPOT4b'
:
True
,
'SPOT5b'
:
True
,
'RE5'
:
False
,
'S2A_full'
:
False
,
# FIXME this changed for S2 in 08/2016
'S2A_full'
:
False
,
'S2A10'
:
False
,
'S2A20'
:
False
,
'S2A60'
:
False
,
...
...
gms_preprocessing/misc/exception_handler.py
View file @
de536db0
...
...
@@ -6,11 +6,14 @@ import shutil
import
sys
import
traceback
import
warnings
from
logging
import
Logger
from
typing
import
Union
,
List
# noqa F401 # flake8 issue
from
..model.gms_object
import
GMS_object
# noqa F401 # flake8 issue
from
..model.gms_object
import
failed_GMS_object
from
..config
import
GMS_config
as
CFG
from
..misc
import
database_tools
as
DB_T
from
.definition_dicts
import
db_jobs_statistics_def
from
.definition_dicts
import
db_jobs_statistics_def
,
proc_chain
__author__
=
'Daniel Scheffler'
...
...
@@ -19,6 +22,8 @@ def trace_unhandled_exceptions(func):
@
functools
.
wraps
(
func
)
def
wrapped_func
(
*
args
,
**
kwargs
):
result
=
None
# noinspection PyBroadException
try
:
result
=
func
(
*
args
,
**
kwargs
)
except
Exception
:
...
...
@@ -29,109 +34,178 @@ def trace_unhandled_exceptions(func):
return
wrapped_func
def
log_uncaught_exceptions
(
GMS_mapper
):
"""Decorator function for handling unexpected exceptions that occurr within GMS mapper functions. Traceback is
sent to logfile of the respective GMS object and the scene ID is added to the 'failed_sceneids' column within
the jobs table of the postgreSQL database.
class
ExceptionHandler
(
object
):
def
__init__
(
self
,
logger
=
None
):
self
.
GMS_objs
=
None
# type: Union[list, dict]
self
.
GMS_mapper_name
=
''
self
.
GMS_mapper_failed
=
False
self
.
_exc_details
=
None
self
.
_logger
=
logger
:param GMS_mapper: A GMS mapper function that takes a GMS object, does some processing and returns it back.
"""
@
property
def
logger
(
self
):
if
not
self
.
_logger
:
self
.
_logger
=
Logger
(
'ExceptionHandler'
,
level
=
CFG
.
job
.
log_level
)
return
self
.
_logger
@
functools
.
wraps
(
GMS_mapper
)
# needed to avoid pickling errors
def
wrapped_GMS_mapper
(
GMS_objs
,
**
kwargs
):
"""
@
logger
.
setter
def
logger
(
self
,
logger
):
self
.
_logger
=
logger
:param GMS_objs: one OR multiple instances of GMS_object or one instance of failed_object
:param kwargs:
:return:
def
__call__
(
self
,
GMS_mapper
):
self
.
log_uncaught_exceptions
(
GMS_mapper
)
def
log_uncaught_exceptions
(
self
,
GMS_mapper
):
"""Decorator function for handling unexpected exceptions that occurr within GMS mapper functions. Traceback is
sent to logfile of the respective GMS object and the scene ID is added to the 'failed_sceneids' column
within the jobs table of the postgreSQL database.
:param GMS_mapper: A GMS mapper function that takes a GMS object, does some processing and returns it back.
"""
try
:
# handle input objects of a GMS mapper that failed in a previous mapper
if
isinstance
(
GMS_objs
,
failed_GMS_object
)
or
\
(
isinstance
(
GMS_objs
,
list
)
and
isinstance
(
GMS_objs
[
0
],
failed_GMS_object
)):
# get a GMS object from which we get the new proc_level
GMS_obj
=
GMS_objs
[
0
]
if
isinstance
(
GMS_objs
,
(
list
,
tuple
))
else
GMS_objs
print
(
"Scene %s (entity ID %s) skipped %s due to an unexpected exception in %s."
%
(
GMS_obj
.
scene_ID
,
GMS_obj
.
entity_ID
,
GMS_mapper
.
__name__
,
GMS_obj
.
failedMapper
))
# TODO should be logged by PC.logger
return
GMS_objs
# in case of just initialized objects:
# update statistics column in jobs table of postgreSQL database to 'started'
if
isinstance
(
GMS_objs
,
collections
.
OrderedDict
)
and
GMS_objs
[
'proc_level'
]
is
None
:
if
not
GMS_objs
[
'subsystem'
]
in
[
'VNIR2'
,
'SWIR'
,
'TIR'
,
'S2A20'
,
'S2A60'
]:
# update statistics column ONLY in case of full cube or first subsystem
DB_T
.
increment_decrement_arrayCol_in_postgreSQLdb
(
CFG
.
job
.
conn_database
,
'jobs'
,
'statistics'
,
cond_dict
=
{
'id'
:
CFG
.
job
.
ID
},
idx_val2decrement
=
db_jobs_statistics_def
[
'started'
]
-
1
,
idx_val2increment
=
db_jobs_statistics_def
[
'started'
])
# run the mapper function and store its results
GMS_objs
=
GMS_mapper
(
GMS_objs
,
**
kwargs
)
# update statistics column in jobs table of postgreSQL database
# get a GMS object from which we get the new proc_level
GMS_obj
=
GMS_objs
[
0
]
if
isinstance
(
GMS_objs
,
(
list
,
tuple
))
else
GMS_objs
# NOTE: in case GMS_obj represents a subsystem and another one has already been marked as FAILED the
# failed_sceneids column and the statistics column is NOT updated once more
# check if another subsystem of the same scene ID already failed - don't increment the stats anymore
if
GMS_obj
.
subsystem
not
in
[
'VNIR2'
,
'SWIR'
,
'TIR'
,
'S2A20'
,
'S2A60'
]:
another_ss_failed
=
False
if
GMS_obj
.
subsystem
:
res
=
DB_T
.
get_info_from_postgreSQLdb
(
CFG
.
job
.
conn_database
,
'jobs'
,
[
'failed_sceneids'
],
{
'id'
:
CFG
.
job
.
ID
})
assert
res
,
"Query delivered no result."
if
res
[
0
][
0
]
is
not
None
and
GMS_obj
.
scene_ID
in
res
[
0
][
0
]:
another_ss_failed
=
True
# update statistics column ONLY in case of full cube or first subsystem and if no other subsystem failed
if
not
another_ss_failed
:
DB_T
.
increment_decrement_arrayCol_in_postgreSQLdb
(
CFG
.
job
.
conn_database
,
'jobs'
,
'statistics'
,
cond_dict
=
{
'id'
:
CFG
.
job
.
ID
},
idx_val2decrement
=
db_jobs_statistics_def
[
GMS_obj
.
proc_level
]
-
1
,
idx_val2increment
=
db_jobs_statistics_def
[
GMS_obj
.
proc_level
])
return
GMS_objs
except
OSError
:
# get Exception details
type_
,
value_
=
sys
.
exc_info
()[:
2
]
traceback_
=
traceback
.
format_exc
()
@
functools
.
wraps
(
GMS_mapper
)
# needed to avoid pickling errors
def
wrapped_GMS_mapper
(
GMS_objs
,
**
kwargs
):
"""
if
value_
.
strerror
==
'Input/output error'
:
# check free disk space
usageNamedTuple
=
shutil
.
disk_usage
(
CFG
.
job
.
path_fileserver
)
percent_free
=
usageNamedTuple
.
free
/
usageNamedTuple
.
total
gigabytes_free
=
usageNamedTuple
.
free
/
(
1024
**
3
)
if
usageNamedTuple
.
free
/
usageNamedTuple
.
total
<
0.025
:
warnings
.
warn
(
'
\n
Catched an unexpected IO error and FREE DISK SPACE IS ONLY %.2f percent '
'(~%.1f GB)!'
%
(
percent_free
*
100
,
gigabytes_free
))
:param GMS_objs: one OR multiple instances of GMS_object or one instance of failed_object
:param kwargs:
:return:
"""
elif
CFG
.
job
.
disable_exception_handler
:
# turn off exception handling and raise the error
raise
self
.
GMS_mapper_name
=
GMS_mapper
.
__name__
self
.
GMS_objs
=
GMS_objs
except
Exception
:
# get Exception details
# noinspection PyBroadException
try
:
self
.
handle_previously_failed
()
self
.
update_progress_started
()
# run the mapper function and store its results
self
.
GMS_objs
=
GMS_mapper
(
GMS_objs
,
**
kwargs
)
self
.
increment_progress
()
return
self
.
GMS_objs
# type: Union[GMS_object, List[GMS_object]]
except
OSError
:
_
,
exc_val
,
_
=
self
.
exc_details
if
exc_val
.
strerror
==
'Input/output error'
:
# check free disk space
usageNamedTuple
=
shutil
.
disk_usage
(
CFG
.
job
.
path_fileserver
)
percent_free
=
usageNamedTuple
.
free
/
usageNamedTuple
.
total
gigabytes_free
=
usageNamedTuple
.
free
/
(
1024
**
3
)
if
usageNamedTuple
.
free
/
usageNamedTuple
.
total
<
0.025
:
self
.
logger
.
warning
(
'
\n
Catched an unexpected IO error and FREE DISK SPACE IS ONLY %.2f percent '
'(~%.1f GB)!'
%
(
percent_free
*
100
,
gigabytes_free
))
elif
CFG
.
job
.
disable_exception_handler
:
raise
else
:
return
self
.
handle_failed
()
# type: failed_GMS_object
except
Exception
:
if
CFG
.
job
.
disable_exception_handler
:
raise
else
:
return
self
.
handle_failed
()
# type: failed_GMS_object
return
wrapped_GMS_mapper
@
property
def
exc_details
(
self
):
if
not
self
.
_exc_details
:
type_
,
value_
=
sys
.
exc_info
()[:
2
]
traceback_
=
traceback
.
format_exc
()
if
CFG
.
job
.
disable_exception_handler
:
# turn off exception handling and raise the error
raise
self
.
_exc_details
=
type_
,
value_
,
traceback_
return
self
.
_exc_details
@
staticmethod
def
is_failed
(
GMS_objs
):
return
isinstance
(
GMS_objs
,
failed_GMS_object
)
or
\
(
isinstance
(
GMS_objs
,
list
)
and
isinstance
(
GMS_objs
[
0
],
failed_GMS_object
))
@
staticmethod
def
get_sample_GMS_obj
(
GMS_objs
):
# type: (Union[list, tuple, collections.OrderedDict]) -> GMS_object
return
\
GMS_objs
if
isinstance
(
GMS_objs
,
collections
.
OrderedDict
)
else
\
GMS_objs
[
0
]
if
isinstance
(
GMS_objs
,
(
list
,
tuple
))
else
GMS_objs
def
handle_previously_failed
(
self
):
if
self
.
is_failed
(
self
.
GMS_objs
):
GMS_obj
=
self
.
get_sample_GMS_obj
(
self
.
GMS_objs
)
# type: failed_GMS_object
print
(
"Scene %s (entity ID %s) skipped %s due to an unexpected exception in %s."
%
(
GMS_obj
.
scene_ID
,
GMS_obj
.
entity_ID
,
self
.
GMS_mapper_name
,
GMS_obj
.
failedMapper
))
# TODO should be logged by PC.logger
return
self
.
GMS_objs
def
update_progress_started
(
self
):
"""in case of just initialized objects:
update statistics column in jobs table of postgreSQL database to 'started'"""
if
isinstance
(
self
.
GMS_objs
,
collections
.
OrderedDict
)
and
self
.
GMS_objs
[
'proc_level'
]
is
None
:
if
not
self
.
GMS_objs
[
'subsystem'
]
or
self
.
GMS_objs
[
'subsystem'
]
in
[
'VNIR1'
,
'S2A10'
]:
self
.
logger
.
debug
(
"Setting job statistics array to 'STARTED'."
)
# update statistics column ONLY in case of full cube or first subsystem
DB_T
.
increment_decrement_arrayCol_in_postgreSQLdb
(
CFG
.
job
.
conn_database
,
'jobs'
,
'statistics'
,
cond_dict
=
{
'id'
:
CFG
.
job
.
ID
},
idx_val2decrement
=
db_jobs_statistics_def
[
'started'
]
-
1
,
idx_val2increment
=
db_jobs_statistics_def
[
'started'
])
def
increment_progress
(
self
):
"""update statistics column in jobs table of postgreSQL database"""
# get a GMS object from which we get the new proc_level
GMS_obj
=
self
.
get_sample_GMS_obj
(
self
.
GMS_objs
)
# NOTE: in case GMS_obj represents a subsystem and another one has already been marked as FAILED the
# failed_sceneids column and the statistics column is NOT updated once more
# check if another subsystem of the same scene ID already failed - don't increment the stats anymore
if
not
GMS_obj
.
subsystem
or
GMS_obj
.
subsystem
in
[
'VNIR1'
,
'S2A10'
]:
another_ss_failed
=
False
if
GMS_obj
.
subsystem
:
# check if another subsystem of the same scene ID has been marked as failed before
res
=
DB_T
.
get_info_from_postgreSQLdb
(
CFG
.
job
.
conn_database
,
'jobs'
,
[
'failed_sceneids'
],
{
'id'
:
CFG
.
job
.
ID
})
assert
res
,
"Query delivered no result."
if
res
[
0
][
0
]
is
not
None
and
GMS_obj
.
scene_ID
in
res
[
0
][
0
]:
self
.
logger
.
debug
(
"Found another failed subsystem of scene %s in the database."
)
another_ss_failed
=
True
# update statistics column ONLY in case of full cube or first subsystem and if no other subsystem failed
if
not
another_ss_failed
:
self
.
logger
.
debug
(
"Decrementing job statistics array for %s objects."
%
proc_chain
[
proc_chain
.
index
(
GMS_obj
.
proc_level
)
-
1
])
self
.
logger
.
debug
(
"Incrementing job statistics array for %s objects."
%
GMS_obj
.
proc_level
)
DB_T
.
increment_decrement_arrayCol_in_postgreSQLdb
(
CFG
.
job
.
conn_database
,
'jobs'
,
'statistics'
,
cond_dict
=
{
'id'
:
CFG
.
job
.
ID
},
idx_val2decrement
=
db_jobs_statistics_def
[
GMS_obj
.
proc_level
]
-
1
,
idx_val2increment
=
db_jobs_statistics_def
[
GMS_obj
.
proc_level
])
@
staticmethod
def
update_progress_failed
(
failed_Obj
):
"""Update statistics column in jobs table of postgreSQL database."""
DB_T
.
increment_decrement_arrayCol_in_postgreSQLdb
(
CFG
.
job
.
conn_database
,
'jobs'
,
'statistics'
,
cond_dict
=
{
'id'
:
CFG
.
job
.
ID
},
idx_val2decrement
=
db_jobs_statistics_def
[
failed_Obj
.
proc_level
],
idx_val2increment
=
db_jobs_statistics_def
[
'FAILED'
])
def
handle_failed
(
self
):
_
,
exc_val
,
exc_tb
=
self
.
exc_details
# collect some informations about failed GMS object and summarize them in failed_GMS_object
failed_Obj
=
\
failed_GMS_object
(
GMS_objs
if
isinstance
(
GMS_objs
,
collections
.
OrderedDict
)
else
GMS_objs
[
0
]
if
isinstance
(
GMS_objs
,
list
)
else
GMS_objs
,
GMS_mapper
.
__name__
,
type_
,
value_
,
traceback_
)
failed_Obj
=
failed_GMS_object
(
self
.
get_sample_GMS_obj
(
self
.
GMS_objs
),
self
.
GMS_mapper_name
,
*
self
.
exc_details
)
# log the exception and raise warning
failed_Obj
.
logger
.
error
(
'
\n
'
+
traceback_
,
exc_info
=
False
)
warnings
.
warn
(
"
\n
Logged an uncaught exception within %s during processing of scene ID %s (entity "
"ID %s):
\n
'%s'
\n
"
%
(
GMS_mapper
.
__name__
,
failed_Obj
.
scene_ID
,
failed_Obj
.
entity_ID
,
value_
))
failed_Obj
.
logger
.
error
(
'
\n
'
+
exc_tb
,
exc_info
=
False
)
self
.
logger
.
warning
(
"
\n
Logged an uncaught exception within %s during processing of scene ID %s "
"(entity ID %s):
\n
'%s'
\n
"
%
(
self
.
GMS_mapper_name
,
failed_Obj
.
scene_ID
,
failed_Obj
.
entity_ID
,
exc_val
))
# add the scene ID to failed_sceneids column in jobs table of DB and update statistics column
# NOTE: in case failed_Obj represents a subsystem and another one has already been marked as FAILED the
...
...
@@ -145,15 +219,14 @@ def log_uncaught_exceptions(GMS_mapper):
DB_T
.
append_item_to_arrayCol_in_postgreSQLdb
(
CFG
.
job
.
conn_database
,
'jobs'
,
{
'failed_sceneids'
:
failed_Obj
.
scene_ID
},
{
'id'
:
CFG
.
job
.
ID
})
# update statistics column in jobs table of postgreSQL database
DB_T
.
increment_decrement_arrayCol_in_postgreSQLdb
(
CFG
.
job
.
conn_database
,
'jobs'
,
'statistics'
,
cond_dict
=
{
'id'
:
CFG
.
job
.
ID
},
idx_val2decrement
=
db_jobs_statistics_def
[
failed_Obj
.
proc_level
],
idx_val2increment
=
db_jobs_statistics_def
[
'FAILED'
])
self
.
update_progress_failed
(
failed_Obj
)
return
failed_Obj
return
wrapped_GMS_mapper
def
log_uncaught_exceptions
(
GMS_mapper
,
logger
=
None
):
exc_handler
=
ExceptionHandler
(
logger
=
logger
)
return
exc_handler
.
log_uncaught_exceptions
(
GMS_mapper
)
def
ignore_warning
(
warning_type
):
...
...
gms_preprocessing/misc/helper_functions.py
View file @
de536db0
...
...
@@ -473,13 +473,20 @@ def scene_ID_to_shapelyPolygon(scene_ID):
"""
Returns a LonLat shapely.Polygon() object corresponding to the given scene_ID.
"""
return
Polygon
(
reorder_CornerLonLat
(
sceneID_to_trueDataCornerLonLat
(
scene_ID
)))
poly
=
Polygon
(
reorder_CornerLonLat
(
sceneID_to_trueDataCornerLonLat
(
scene_ID
)))
if
not
poly
.
is_valid
:
poly
=
poly
.
buffer
(
0
)
assert
poly
.
is_valid
return
poly
def
CornerLonLat_to_shapelyPoly
(
CornerLonLat
):
"""Returns a shapely.Polygon() object based on the given coordinate list. """
return
Polygon
(
reorder_CornerLonLat
(
CornerLonLat
))
poly
=
Polygon
(
reorder_CornerLonLat
(
CornerLonLat
))
if
not
poly
.
is_valid
:
poly
=
poly
.
buffer
(
0
)
assert
poly
.
is_valid
return
poly
def
find_in_xml_root
(
namespace
,
xml_root
,
branch
,
*
branches
,
findall
=
None
):
...
...
gms_preprocessing/misc/path_generator.py
View file @
de536db0
...
...
@@ -5,6 +5,7 @@ import re
import
tempfile
import
warnings
import
uuid
from
logging
import
Logger
from
..config
import
GMS_config
as
CFG
from
.definition_dicts
import
get_GMS_sensorcode
...
...
@@ -336,7 +337,8 @@ def get_path_ac_options(GMS_identifier):
assert
os
.
path
.
exists
(
path_ac
)