Commit 451ba2ec authored by cponcele's avatar cponcele
Browse files

add sonar_netcdf_validator

parent 2571ead4
import inspect
import re
import netCDF4 as nc
import sonar_netcdf.utils.print_color as p
import sonar_netcdf.sonar_validator_model as nc_model
class SonarChecker:
def __init__(self):
# parse static model files
self.classes = {}
for name, obj in inspect.getmembers(nc_model):
if inspect.isclass(obj):
self.classes[name] = obj
self.current_file_path = ""
self._reset_stats()
def _head(self, msg):
p.header(msg)
def _info(self, msg):
p.info(msg)
def _warning(self, msg):
self.warning_count += 1
p.warning(msg)
# logging.warning(msg)
def _error(self, msg):
self.error_count += 1
p.error(msg)
# logging.error(msg)
def _get_validator_class(self, path: str):
"""Return the associated test class in sonar_validator_model for this group, or None if not found"""
if not path.endswith("/"):
path = path + "/"
for name in self.classes:
pattern = self.classes[name].get_group_path_pattern()
match = re.fullmatch(pattern, path)
if match:
return self.classes[name]
return None
def validate_grp(self, dataset: nc.Dataset):
self._info(f"Checking group {dataset.path}")
validator_class = self._get_validator_class(dataset.path)
if validator_class is None:
self._warning(f"Cannot match any known group definition for {dataset.path}")
return
# vlen type
for expected_type in validator_class._vlen_types:
if expected_type not in dataset.vltypes:
self._error(f"Missing vlen type declaration {expected_type} expected in {dataset.path}")
for expected_type in validator_class._enum_types:
if expected_type not in dataset.vltypes:
self._error(f"Missing enum declaration {expected_type} expected in {dataset.path}")
for expected_type in validator_class._vlen_types:
if expected_type not in dataset.enumtypes:
self._error(f"Missing vlen type declaration {expected_type} expected in {dataset.path}")
# compound types not done
# enum
# Dimensions
# first validate dimensions, check that all dimension are defined in source dataset
for expected_dimension in validator_class._dimensions:
if expected_dimension not in dataset.dimensions:
self._error(f"Missing dimension {expected_dimension} in {dataset.path}")
# Attributes
for expected in validator_class._attributes:
if expected not in dataset.ncattrs():
self._error(f"Missing group Attribute {expected} in {dataset.path}")
# Coordinate Variables
for expected in validator_class._coordinate_variables:
if expected not in dataset.variables:
self._error(f"Missing mandatory coordinate variable {expected} in {dataset.path}")
# Variables
for expected in validator_class._mandatory_variables:
if expected not in dataset.variables:
self._error(f"Missing mandatory variable {expected} in {dataset.path}")
for expected in validator_class._recommanded_variables:
if expected not in dataset.variables:
self._warning(f"Recommanded variable {expected} is missing in {dataset.path} , consider to add it")
# we do not do any check on mandatory_if_applicable variables or optional variable
# recurse subgroups
for grp in dataset.groups:
self.validate_grp(dataset[grp])
def validate_file(self, file_path: str):
self._reset_stats()
self.current_file_path = file_path
self._head(f"Checking file f{file_path}")
# open the file
with nc.Dataset(file_path, mode="r") as file:
# and validate each group
self.validate_grp(file)
self._print_result()
def _reset_stats(self):
self.error_count = 0
self.warning_count = 0
def _print_result(self):
self._head(f"{self.current_file_path} : {self.error_count} errors, {self.warning_count} warnings")
if __name__ == "__main__":
filename = "c:/data/datasets/netcdfInspector/PELGAS19_012_20190507_042828_01_1.xsf.nc"
checker= SonarChecker()
checker.validate_file(filename)
checker.validate_file("c:/data/datasets/netcdfInspector/GLOBE_MERGED_GAZCOGNE3_003_20180829_061940_1.xsf_TO_GAZCOGNE3_003_20180829_070941_1.xsf_v783u7_w_ES333-7C-333000.nc")
checker.validate_file("C:\data\datasets\SonarNetcdf\HYDROMOMAR-D20200904-T093757.nc")
"""
Utility class generated see for Sonar-netcdf see https://gitlab.ifremer.fr/fleet/formats/pySonar-netcdf
"""
import netCDF4 as nc
import numpy as np
# pylint: disable=E1101
# pylint: disable=R0904
# pylint: disable=C0305
# pylint: disable=C0302
class RootGrpValidator:
@staticmethod
def get_group_path_pattern():
return "/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=["Conventions","date_created","keywords","license","rights","sonar_convention_authority","sonar_convention_name","sonar_convention_version","summary","title","processing_status","xsf_convention_version"]
#Dimensions
_dimensions=["history"]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=[]
_optional_variables=["crs","history"]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class EnvironmentGrpValidator:
@staticmethod
def get_group_path_pattern():
return RootGrpValidator.get_group_path_pattern()+"Environment/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=[]
#Dimensions
_dimensions=["frequency"]
#Variables
_coordinate_variables=["frequency"]
#Variables
_mandatory_variables=["absorption_indicative","sound_speed_indicative"]
_optional_variables=[]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class SoundSpeedProfileGrpValidator:
@staticmethod
def get_group_path_pattern():
return EnvironmentGrpValidator.get_group_path_pattern()+"Sound_speed_profile/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=[]
#Dimensions
_dimensions=["profile","sampleCount"]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=["time","depth","sample_count","sound_speed","salinity"]
_optional_variables=["profile","lat","lon","temperature","absorption"]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class SurfaceSoundSpeedGrpValidator:
@staticmethod
def get_group_path_pattern():
return EnvironmentGrpValidator.get_group_path_pattern()+"Surface_sound_speed/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=[]
#Dimensions
_dimensions=["time"]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=["surface_sound_speed","time"]
_optional_variables=[]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class TideGrpValidator:
@staticmethod
def get_group_path_pattern():
return EnvironmentGrpValidator.get_group_path_pattern()+"Tide/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=[]
#Dimensions
_dimensions=["time"]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=["tide_indicative","time"]
_optional_variables=[]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class DynamicDraughtGrpValidator:
@staticmethod
def get_group_path_pattern():
return EnvironmentGrpValidator.get_group_path_pattern()+"Dynamic_draught/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=[]
#Dimensions
_dimensions=["time"]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=["draught_indicative","time"]
_optional_variables=[]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
class PlatformGrpValidator:
@staticmethod
def get_group_path_pattern():
return RootGrpValidator.get_group_path_pattern()+"Platform/"
_enum_types=["transducer_type_t"]
_vlen_types=[]
#Attributes
_attributes=["platform_code_ICES","platform_name","platform_type"]
#Dimensions
_dimensions=["transducer","position","MRU"]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=["MRU_ids","position_ids","transducer_ids","transducer_function"]
_optional_variables=[]
_recommanded_variables=["MRU_offset_x","MRU_offset_y","MRU_offset_z","MRU_rotation_x","MRU_rotation_y","MRU_rotation_z","position_offset_x","position_offset_y","position_offset_z","transducer_offset_x","transducer_offset_y","transducer_offset_z","transducer_rotation_x","transducer_rotation_y","transducer_rotation_z","water_level"]
_mandatory_if_applicable_variables=[]
#Sub groups
class PositionGrpValidator:
@staticmethod
def get_group_path_pattern():
return PlatformGrpValidator.get_group_path_pattern()+"Position/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=[]
#Dimensions
_dimensions=[]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=[]
_optional_variables=[]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class PositionSubGroupValidator:
@staticmethod
def get_group_path_pattern():
return PositionGrpValidator.get_group_path_pattern()+".*"+"/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=["description"]
#Dimensions
_dimensions=["time"]
#Variables
_coordinate_variables=["time"]
#Variables
_mandatory_variables=["latitude","longitude","heading","speed_over_ground","speed_relative","height_above_reference_ellipsoid","altitude"]
_optional_variables=["course_over_ground","distance"]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class NmeaGrpValidator:
@staticmethod
def get_group_path_pattern():
return PositionSubGroupValidator.get_group_path_pattern()+"NMEA/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=["description"]
#Dimensions
_dimensions=["time"]
#Variables
_coordinate_variables=["time"]
#Variables
_mandatory_variables=[]
_optional_variables=["NMEA_datagram"]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
class AttitudeGrpValidator:
@staticmethod
def get_group_path_pattern():
return PlatformGrpValidator.get_group_path_pattern()+"Attitude/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=[]
#Dimensions
_dimensions=[]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=[]
_optional_variables=[]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class AttitudeSubGroupValidator:
@staticmethod
def get_group_path_pattern():
return AttitudeGrpValidator.get_group_path_pattern()+".*"+"/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=["description"]
#Dimensions
_dimensions=["time"]
#Variables
_coordinate_variables=["time"]
#Variables
_mandatory_variables=["heading","heading_rate","pitch","roll","vertical_offset"]
_optional_variables=["pitch_rate","roll_rate"]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class ProvenanceGrpValidator:
@staticmethod
def get_group_path_pattern():
return RootGrpValidator.get_group_path_pattern()+"Provenance/"
_enum_types=[]
_vlen_types=[]
#Attributes
_attributes=["conversion_software_name","conversion_software_version","conversion_time"]
#Dimensions
_dimensions=["filenames"]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=["source_filenames"]
_optional_variables=[]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
class SonarGrpValidator:
@staticmethod
def get_group_path_pattern():
return RootGrpValidator.get_group_path_pattern()+"Sonar/"
_enum_types=["beam_stabilisation_t","beam_t","conversion_equation_t","transmit_t"]
_vlen_types=[]
#Attributes
_attributes=["sonar_manufacturer","sonar_model","sonar_serial_number","sonar_software_name","sonar_software_version","sonar_type"]
#Dimensions
_dimensions=[]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=[]
_optional_variables=[]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class BeamGroup1GrpValidator:
@staticmethod
def get_group_path_pattern():
return SonarGrpValidator.get_group_path_pattern()+"Beam_group[0-9]*"+"/"
_enum_types=[]
_vlen_types=["sample_t","angle_t"]
#Attributes
_attributes=["beam_mode","conversion_equation_type","preferred_MRU","preferred_position"]
#Dimensions
_dimensions=["beam","subbeam","ping_time","tx_beam"]
#Variables
_coordinate_variables=["beam","ping_time","mean_time"]
#Variables
_mandatory_variables=["backscatter_i","backscatter_r","echoangle_major","echoangle_minor","echoangle_major_sensitivity","echoangle_minor_sensitivity","beamwidth_receive_major","beamwidth_receive_minor","beamwidth_transmit_major","beamwidth_transmit_minor","rx_beam_rotation_phi","rx_beam_rotation_theta","rx_beam_rotation_psi","tx_beam_rotation_phi","tx_beam_rotation_theta","tx_beam_rotation_psi","beam_stabilisation","beam_type","equivalent_beam_angle","gain_correction","non_quantitative_processing","receiver_sensitivity","sample_interval","sample_time_offset","blanking_interval","time_varied_gain","transducer_gain","transmit_duration_nominal","receive_duration_effective","transmit_frequency_start","transmit_frequency_stop","transmit_power","transmit_source_level","transmit_type","receive_transducer_index","transmit_transducer_index","transmit_beam_index","active_MRU","active_position_sensor","platform_latitude","platform_longitude","platform_heading","platform_pitch","platform_roll","platform_vertical_offset"]
_optional_variables=["detected_bottom_range","transmit_bandwidth","sound_speed_at_transducer","tx_transducer_depth","waterline_to_chart_datum"]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class BathymetryGrpValidator:
@staticmethod
def get_group_path_pattern():
return BeamGroup1GrpValidator.get_group_path_pattern()+"Bathymetry/"
_enum_types=["detection_type_t","seafloor_backscatter_equation_t"]
_vlen_types=["seabed_image_sample_t"]
#Attributes
_attributes=["detection_conversion_equation_type"]
#Dimensions
_dimensions=["detection"]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=["detection_x","detection_y","detection_z","status","status_detail","detection_type","detection_rx_transducer_index","detection_tx_beam","detection_tx_transducer_index","detection_backscatter_r","detection_backscatter_i","detection_source_level_applied","detection_receiver_sensitivity_applied","detection_backscatter_calibration","detection_time_varying_gain","detection_mean_absorption_coefficient","seabed_image_samples_r"]
_optional_variables=["detection_latitude","detection_longitude","detection_beam_pointing_angle","detection_two_way_travel_time","detection_quality_factor","detection_rx_beam","seabed_image_start_range","seabed_image_center"]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
#Sub groups
class GridGroup1GrpValidator:
@staticmethod
def get_group_path_pattern():
return SonarGrpValidator.get_group_path_pattern()+"Grid_group[0-9]*"+"/"
_enum_types=["backscatter_type_t","range_axis_interval_type_t","ping_axis_interval_type_t"]
_vlen_types=[]
#Attributes
_attributes=["beam_mode","conversion_equation_type"]
#Dimensions
_dimensions=["beam","tx_beam","frequency","ping_axis","range_axis"]
#Variables
_coordinate_variables=[]
#Variables
_mandatory_variables=["beam","frequency","beam_reference","backscatter_type","ping_axis_interval_type","ping_axis_interval_value","range_axis_interval_type","range_axis_interval_value","cell_ping_time","integrated_backscatter","beamwidth_receive_major","beamwidth_receive_minor","beamwidth_transmit_major","beamwidth_transmit_minor","rx_beam_rotation_phi","rx_beam_rotation_theta","rx_beam_rotation_psi","tx_beam_rotation_phi","tx_beam_rotation_theta","tx_beam_rotation_psi","beam_stabilisation","beam_type","equivalent_beam_angle","gain_correction","non_quantitative_processing","receiver_sensitivity","sample_interval","sample_time_offset","blanking_interval","time_varied_gain","transducer_gain","transmit_duration_nominal","receive_duration_effective","transmit_frequency_start","transmit_frequency_stop","transmit_power","transmit_source_level","transmit_type","receive_transducer_index","cell_latitude","cell_longitude","cell_depth"]
_optional_variables=["detected_bottom_range","transmit_bandwidth","sound_speed_at_transducer","tx_transducer_depth","waterline_to_chart_datum"]
_recommanded_variables=[]
_mandatory_if_applicable_variables=[]
......@@ -7,46 +7,9 @@ import matplotlib.pyplot as plt
# ensure minimum size for figures
plt.rcParams["figure.dpi"] = 270
from sonar_netcdf.utils.print_color import header,warning,error,pprint
#pylint:disable=redefined-outer-name
class bcolors:
"""
Utility for color display in terminal or jupyter
print(f"{bcolors.WARNING}{text}{bcolors.ENDC}")
"""
HEADER = "\033[95m"
OKBLUE = "\033[94m"
OKCYAN = "\033[96m"
OKGREEN = "\033[92m"
WARNING = "\033[93m"
FAIL = "\033[91m"
ENDC = "\033[0m"
BOLD = "\033[1m"
UNDERLINE = "\033[4m"
def pprint(msg, back_ground_color=None):
"""Intercept print call"""
if back_ground_color is None:
print(msg)
else:
print(f"{back_ground_color}{msg}{bcolors.ENDC}")
def error(msg: str):
pprint(msg, back_ground_color=bcolors.FAIL)
def warning(msg: str):
pprint(msg, back_ground_color=bcolors.WARNING)
def header(msg: str):
pprint(msg, back_ground_color=bcolors.BOLD)
class NcReader:
"""
Class used for reading a Sonar-netcdf like file
......
class bcolors:
"""
Utility for color display in terminal or jupyter
print(f"{bcolors.WARNING}{text}{bcolors.ENDC}")
"""
HEADER = "\033[95m"
OKBLUE = "\033[94m"
OKCYAN = "\033[96m"
OKGREEN = "\033[92m"
WARNING = "\033[96m"
FAIL = "\033[91m"
ENDC = "\033[0m"
BOLD = "\033[1m"
UNDERLINE = "\033[4m"
def info(msg):
pprint(msg)
def pprint(msg, back_ground_color=None):
"""Intercept print call"""
if back_ground_color is None:
print(msg)
else:
print(f"{back_ground_color}{msg}{bcolors.ENDC}")
def error(msg: str):
pprint(msg, back_ground_color=bcolors.FAIL)
def warning(msg: str):
pprint(msg, back_ground_color=bcolors.WARNING)
def header(msg: str):
pprint(msg, back_ground_color=bcolors.BOLD)
Markdown is supported
0%