Source code for gdgps_apps.defines

|APPS| makes use of many enumerations throughout its infrastructure. The defines module provides pythonic access to the
definitions of those enumerations as well as to convenience methods for normalizing short and long forms in a case
insensitive way. It is a best practice when using the library never to use the raw enumeration values directly as
literals but rather to reference them through the defines module. This both makes code more readable and more resilient
to library updates.

import pytz

[docs]def lower(chc): """ For choices empty string and None should be treated the same, never use empty string, always use None in choice lists. :param chc: :return: None if empty string or None, else return the lower case version of the string """ if chc is None or chc == "": return None return chc.lower()
[docs]def fetch_choice(choice, choices, idx): c = lower(choice) for chc in choices: if chc[0] is not None: if c == chc[0].lower(): return chc[idx] elif c == chc[1].lower(): return chc[idx] elif c is None: return chc[idx] if chc[0] == 'M' and chc[1] == 'Mixed' and c == ' ': return 'M' raise KeyError('%s is not a valid choice among: %s' % (choice, choices))
[docs]def rank_choices(left, right, choices): if left == right: return 0 left_l = lower(left) right_l = lower(right) seen_l = False seen_r = False for choice in choices: if left_l == lower(choice[0]) or left_l == lower(choice[1]): seen_l = True if right_l == lower(choice[0]) or right_l == lower(choice[1]): seen_r = True if seen_l or seen_r: break if seen_l: if seen_r: return 0 else: return -1 else: if seen_r: return 1 else: return None
[docs]def get_verbose(abbr, choices): return fetch_choice(abbr, choices, 1)
[docs]def normalize(choice, choices): return fetch_choice(choice, choices, 0)
[docs]class DataFlag(object): """ DataFlags are issued against submissions and exist at one of four defined levels. Similar to ubiquitous log level conventions, they indicate the degree to which the user's attention is required to resolve an issue if one exists. """ SUCCESS = 'S' """ Indicates that some operation was successful. """ INFO = 'I' """ Includes some information that may or may not be of interest to the user. """ WARN = 'W' """ Indicates that |APPS| should be able, or was able to process the submission but there may be some caveats that the user should be aware of before reviewing the results. """ ERROR = 'E' """ Indicates that there was an issue with the submission that is unresolvable without additional intervention from the user. """ LEVELS = ( (SUCCESS, 'Success'), (INFO, 'Info'), (WARN, 'Warning'), (ERROR, 'Error') ) """ The valid list of level definitions including short form formats and their long form counterparts. """
[docs] @classmethod def get_verbose_level(cls, level): """ Convert the given level enumeration string into its canonical long form format :param level: A valid level enumeration, case insensitve :type level: str :return: The long form format of the level :raises ValueError: if the level enumeration parameter does not refer to a valid level """ return get_verbose(level, cls.LEVELS)
[docs] @classmethod def normalize_level(cls, level): """ Convert the given level enumeration string into its canonical short form format :param level: A valid level enumeration, case insensitive :type level: str :return: The short form format of the level :raises ValueError: if the level enumeration parameter does not refer to a valid level """ return normalize(level, cls.LEVELS)
[docs]class Data(object): NASCENT = 'N' SUBMITTED = 'S' VERIFIED = 'V' APPROVED = 'O' WAITING = 'W' QUEUED = 'Q' ERROR = 'E' PROCESSING = 'P' DONE = 'D' AVAILABLE = 'A' RETRIEVED = 'R' STATES = ( (NASCENT, 'Nascent'), (SUBMITTED, 'Submitted'), (VERIFIED, 'Verified'), (APPROVED, 'Approved'), (WAITING, 'Waiting'), (QUEUED, 'Queued'), (ERROR, 'Error'), (PROCESSING, 'Processing'), (DONE, 'Done'), (AVAILABLE, 'Available'), (RETRIEVED, 'Retrieved') )
[docs] @classmethod def get_verbose_state(cls, state): return get_verbose(state, cls.STATES)
[docs] @classmethod def normalize_state(cls, state): return normalize(state, cls.STATES)
[docs] @classmethod def rank_state(cls, left, right): return rank_choices(left, right, cls.STATES)
PRIVATE = 'X' PUBLIC = 'A' ACCESS_CHOICES = ( (PRIVATE, 'Private'), (PUBLIC, 'Public') )
[docs] @classmethod def get_verbose_access(cls, setting): return get_verbose(setting, cls.ACCESS)
[docs] @classmethod def normalize_access(cls, setting): return normalize(setting, cls.ACCESS)
ACCESS_HELP = 'Restrict or allow access to the data. Public means anyone with the link can view/download ' \ 'source files and results of this data submission, but will not be able to make any ' \ 'changes. Private, means only the owner will be able to view the data.' ACCESS_DEFAULT = PRIVATE NAME_HELP = 'The name to use for the submission. If no name is provided the file name will be used by default.' EMAIL_NOTIFY_CHOICES = ( (None, 'Account Default'), (False, 'No'), (True, 'Yes') ) EMAIL_NOTIFY_DEFAULT = None EMAIL_NOTIFY_HELP = 'Send an email once processing has completed.' _name_notify = ['name', 'email_notify', 'access'] OWNER_UPDATE_TABLE = { SUBMITTED: {}, VERIFIED: {'fields': _name_notify + ['state'], 'state': (APPROVED,)}, WAITING: {'fields': _name_notify}, APPROVED: {'fields': _name_notify}, QUEUED: {'fields': _name_notify}, ERROR: {'fields': _name_notify}, PROCESSING: {'fields': _name_notify}, DONE: {'fields': ['name', 'access']}, AVAILABLE: {'fields': ['name', 'access']}, RETRIEVED: {'fields': ['name', 'access']}, } # can only change sources in one of these states, a source change in any of these states kicks the # data back to SUBMITTED SOURCE_CHANGE_STATES = [NASCENT, VERIFIED, WAITING, ERROR] SOURCE_DELETE_STATES = SOURCE_CHANGE_STATES + [DONE, AVAILABLE, RETRIEVED]
[docs]class Storage(object): GZIP = 'gz' BZIP2 = 'bz2' TAR = 'tar' ZIP = 'zip' LZMA = 'xz' UNIX = 'Z'
[docs]class UserProfile(object): GZIP = Storage.GZIP BZIP2 = Storage.BZIP2 UNIX = Storage.UNIX ZIP = Storage.ZIP TAR = Storage.TAR COMPRESSION_PREFERENCES = ( (GZIP, GZIP), (BZIP2, BZIP2), (UNIX, UNIX) ) COMPRESSION_PREFERENCE_DEFAULT = GZIP COMPRESSION_PREFERENCE_HELP = 'Result files will be compressed using this format.' ARCHIVE_PREFERENCES = ( (TAR, TAR), (ZIP, ZIP) ) ARCHIVE_PREFERENCE_DEFAULT = TAR ARCHIVE_PREFERENCE_HELP = 'Result files will be aggregated and archived using this format.' EMAIL_NOTIFY_DEFAULT = False EMAIL_NOTIFY_HELP = 'Send an email once submitted GNSS data has been processed and results are available. This ' \ 'can be overridden on individual submissions.' NO_PROMPT_PROFESSING_DEFAULT = False NO_PROMPT_PROCESSING_HELP = 'Do not wait for user review and authorization, once data has been verified, just ' \ 'automatically process it.' KEEP_SOURCE_FILES_DEFAULT = False KEEP_SOURCE_FILES_HELP = 'Retain source data after processing has completed. Source data will be retrievable.' TIMEZONES = tuple(zip(pytz.all_timezones, pytz.all_timezones)) TIMEZONE_DEFAULT = 'UTC' TIMEZONE_HELP = 'The timezone that all times, other than GNSS times will be output in.'
[docs]class OrbitClockProduct(object): FINAL = 'F' RAPID = 'R' ULTRA = 'U' REAL_TIME = 'T' # The order these appear in the products tuple is important! # It's used to rank the products by precision, they are listed here in increasing order of precision PRODUCTS = ( (REAL_TIME, 'Real-Time'), (ULTRA, 'Ultra'), (RAPID, 'Rapid'), (FINAL, 'Final'), )
[docs] @classmethod def get_verbose_product(cls, product): return get_verbose(product, cls.PRODUCTS)
[docs] @classmethod def normalize_product(cls, product): return normalize(product, cls.PRODUCTS)
[docs] @classmethod def product_rank(cls, product): p = product.lower() idx = 0 for product in cls.PRODUCTS: if p == product[0].lower() or p == product[1].lower(): return idx idx += 1 raise KeyError('%s is not a valid choice among: %s' % (product, cls.PRODUCTS))
[docs]class GIPSYData(Data, OrbitClockProduct): # Options UNRESOLVED = 'Un' STATIC = 'S' KINEMATIC = 'K' # deprecated, but leave in until all code can be updated KINEMATIC_SLOW = 'K' KINEMATIC_FAST = 'F' PROCESSING_MODES = ( (STATIC, 'Static'), (KINEMATIC_SLOW, 'Kinematic-Slow'), (KINEMATIC_FAST, 'Kinematic-Fast') ) PROCESSING_MODE_DEFAULT = STATIC PROCESSING_MODE_HELP = 'Compute either a stationary, %s (%s) solution, or a track, %s (%s) solution of a slow ' \ 'moving receiver that moves less than few meters/day or a %s (%s) fast moving receiver.' % ( get_verbose(STATIC, PROCESSING_MODES), STATIC, get_verbose(KINEMATIC_SLOW, PROCESSING_MODES), KINEMATIC_SLOW, get_verbose(KINEMATIC_FAST, PROCESSING_MODES), KINEMATIC_FAST )
[docs] @classmethod def get_verbose_processing_mode(cls, processing_mode): return get_verbose(processing_mode, cls.PROCESSING_MODES)
[docs] @classmethod def normalize_processing_mode(cls, processing_mode): return normalize(processing_mode, cls.PROCESSING_MODES)
[docs] @classmethod def get_verbose_source_type(cls, source_type): return get_verbose(source_type, cls.SOURCE_TYPES)
[docs] @classmethod def normalize_source_type(cls, source_type): return normalize(source_type, cls.SOURCE_TYPES)
ARCHIVE_FILE = 'A' SUMMARY_FILE = 'S' TDP_FILE = 'T' SOLUTION_FILE = 'P' GDCOV_FILE = 'G' DETAIL_FILE = 'D' KML_FILE = 'K' RESULT_TYPES = ( (ARCHIVE_FILE, 'Archive'), (SUMMARY_FILE, 'Summary'), (TDP_FILE, 'Time Dependent Parameters'), (SOLUTION_FILE, 'Solution'), (GDCOV_FILE, 'Covariance'), (DETAIL_FILE, 'Detail'), (KML_FILE, 'Keyhole Markup Language') )
[docs] @classmethod def get_verbose_result_type(cls, result_type): return get_verbose(result_type, cls.RESULT_TYPES)
[docs] @classmethod def normalize_result_type(cls, result_type): return normalize(result_type, cls.RESULT_TYPES)
RESULT_TYPES_DEFAULT = None RESULT_TYPES_HELP = 'The type of data result the file is, either ' + ', or'.join( ['%s (%s)' % (s, v) for s, v in RESULT_TYPES] ) TROP_OFF = 'OFF' TROP_GMF = 'GMF' TROP_VMF1 = 'VMF1' TROP_GPT2 = 'GPT2' TROP_PROVIDED = 'PRV' TROPOSPHERE_MODELS = ( (TROP_OFF, 'Off'), (TROP_GMF, 'GMF'), (TROP_VMF1, 'VMF1'), (TROP_GPT2, 'GPT2'), (TROP_PROVIDED, 'Provided') ) TROPOSPHERE_MODEL_HELP = 'Select which of the following standard models to use to model the tropospheric delay. ' \ 'If the receiver was in low earth orbit, or otherwise above the troposphere you should ' \ 'turn troposphere modeling off. To upload your own pressure data select Provided. VMF1 ' \ 'can only work for data older than several days. Static runs will default to GMF, ' \ 'Kinematic runs will default to GPT2.'
[docs] @classmethod def get_verbose_troposphere_model(cls, trop_model): return get_verbose(trop_model, cls.TROPOSPHERE_MODELS)
[docs] @classmethod def normalize_troposphere_model(cls, trop_model): return normalize(trop_model, cls.TROPOSPHERE_MODELS)
ANTENNA_CALIBRATION_HELP = 'User supplied antenna calibration file. Format should follow IGS standard.' MODEL_PRESSURE_HELP = 'This option is deprecated, set troposphere model to `Provided` instead.' OCEAN_LOADING_HELP = 'Model the effect of the ocean loading on crust deformation. This should be enabled for ' \ 'receivers on the ground, and especially near coastlines.' MODEL_TIDES_HELP = 'Use polar and solid earth tidal models in the solution. If the receiver was on the ' \ 'ground this should be enabled, if it was airborne this should be disabled.' SINGLE_FREQUENCY = 'S' DUAL_FREQUENCY = 'D' MEASUREMENT_TYPES = ( (SINGLE_FREQUENCY, 'Single Frequency'), (DUAL_FREQUENCY, 'Dual Frequency'), (UNRESOLVED, 'Unresolved') )
[docs] @classmethod def get_verbose_measurement_type(cls, measurement_type): return get_verbose(measurement_type, cls.MEASUREMENT_TYPES)
[docs] @classmethod def normalize_measurement_type(cls, measurement_type): return normalize(measurement_type, cls.MEASUREMENT_TYPES)
FLAT = 'FLAT' SINE = 'SIN' ROOT_SINE = 'SQRTSIN' ELEV_DEP_WEIGHTINGS = ( (FLAT, 'Flat'), (SINE, 'Sin'), (ROOT_SINE, 'SqrtSin') ) ELEV_DEP_WEIGHTING_HELP = 'Elevation dependent weighting scheme in the receiver\'s local frame. The sigma used ' \ 'for the measurement will be computed as: Flat: DataSigma, ' \ 'Sin: DataSigma / sin(elevation), SqrtSin: DataSigma / sqrt( sin(elevation) ). ' \ 'For Sin and SqrtSin, elevations below 5 deg are treated as 5 deg.'
[docs] @classmethod def get_verbose_elev_dep_weighting(cls, elev_dep_weighting): return get_verbose(elev_dep_weighting, cls.ELEV_DEP_WEIGHTINGS)
[docs] @classmethod def normalize_elev_dep_weighting(cls, elev_dep_weighting): return normalize(elev_dep_weighting, cls.ELEV_DEP_WEIGHTINGS)
# positioning preview results BEST = 'B' PRODUCTS = [(BEST, 'Best Available'), ] + list(OrbitClockProduct.PRODUCTS) PRODUCT_DEFAULT = BEST PRODUCT_HELP = ( 'The GNSS satellite orbit and clock state data product to use to compute the solution. In increasing ' 'order of precision the products are %s (%s), %s (%s), %s (%s), %s (%s). If %s (%s) is selected, the ' 'most precise product currently available during the time frame of the measurement will be used. %s ' 'products are available immediately, %s products are available within about 2 hours, %s products are ' 'available the next day and %s products are usually available within two weeks.' % ( OrbitClockProduct.get_verbose_product(OrbitClockProduct.REAL_TIME), OrbitClockProduct.REAL_TIME, OrbitClockProduct.get_verbose_product(OrbitClockProduct.ULTRA), OrbitClockProduct.ULTRA, OrbitClockProduct.get_verbose_product(OrbitClockProduct.RAPID), OrbitClockProduct.RAPID, OrbitClockProduct.get_verbose_product(OrbitClockProduct.FINAL), OrbitClockProduct.FINAL, get_verbose(BEST, PRODUCTS), BEST, OrbitClockProduct.get_verbose_product(OrbitClockProduct.REAL_TIME), OrbitClockProduct.get_verbose_product(OrbitClockProduct.ULTRA), OrbitClockProduct.get_verbose_product(OrbitClockProduct.RAPID), OrbitClockProduct.get_verbose_product(OrbitClockProduct.FINAL), ) ) ELEV_ANGLE_CUTOFF_HELP = 'Signals received below this angle will be excluded from the calculation. This can be ' \ 'helpful in reducing multi-path error.' ELEV_ANGLE_CUTOFF_MIN = -90.0 ELEV_ANGLE_CUTOFF_MAX = 90.0 SOLUTION_PERIOD_MIN = 1 SOLUTION_PERIOD_HELP = 'The period in seconds at which to compute a positioning solution. If no period is ' \ 'specified APPS will attempt to use the data rate of the file.' GENERATE_QUATERNIONS_HELP = 'When processing in kinematic mode estimate attitude quaternions automatically from ' \ 'velocity vector.' _name_notify = ['name', 'email_notify', 'access'] GIPSY_FIELDS = [ 'access', 'processing_mode', 'product', #'codes', 'troposphere_model', #'model_pressure', DEPRECATED - SET troposphere_model = PROVIDED to upload pressure files 'ocean_loading', 'model_tides', 'elev_dep_weighting', 'elev_angle_cutoff', 'solution_period', 'generate_quaternions' ] ANCILLARY_SOURCE_FIELDS = [ get_verbose(ANTENNA_CALIBRATION_FILE, SOURCE_TYPES).lower().replace(' ', '_'), get_verbose(PRESSURE_FILE, SOURCE_TYPES).lower(), get_verbose(ATTITUDE_FILE, SOURCE_TYPES).lower() ] SUBMISSION_FIELDS = _name_notify + GIPSY_FIELDS + ANCILLARY_SOURCE_FIELDS OWNER_UPDATE_TABLE = { Data.NASCENT: {'fields': SUBMISSION_FIELDS + [get_verbose(RINEX_FILE, SOURCE_TYPES).lower()]}, Data.SUBMITTED: {}, Data.VERIFIED: {'fields': SUBMISSION_FIELDS + ['state'], 'state': (Data.APPROVED,), 'transition': {Data.SUBMITTED: GIPSY_FIELDS + ANCILLARY_SOURCE_FIELDS}}, Data.WAITING: {'fields': SUBMISSION_FIELDS, 'transition': {Data.SUBMITTED: GIPSY_FIELDS + ANCILLARY_SOURCE_FIELDS}}, Data.APPROVED: {'fields': _name_notify}, Data.QUEUED: {'fields': _name_notify}, Data.ERROR: {'fields': SUBMISSION_FIELDS, 'transition': {Data.SUBMITTED: GIPSY_FIELDS + ANCILLARY_SOURCE_FIELDS}}, Data.PROCESSING: {'fields': _name_notify}, Data.DONE: {'fields': ['name', 'access']}, Data.AVAILABLE: {'fields': ['name', 'access']}, Data.RETRIEVED: {'fields': ['name', 'access']}, } PROCESSING_MODE_SETTINGS = { STATIC: { 'solution_period': 300, 'troposphere_model': TROP_GMF, 'ocean_loading': True, 'model_tides': True, 'elev_angle_cutoff': 7.5, 'elev_dep_weighting': ROOT_SINE, 'generate_quaternions': False }, KINEMATIC_SLOW: { 'solution_period': None, 'troposphere_model': TROP_GPT2, 'ocean_loading': True, 'model_tides': True, 'elev_angle_cutoff': 7.5, 'elev_dep_weighting': SINE, 'generate_quaternions': False }, KINEMATIC_FAST: { 'solution_period': None, 'troposphere_model': TROP_GPT2, 'ocean_loading': False, 'model_tides': False, 'elev_angle_cutoff': 0, 'elev_dep_weighting': ROOT_SINE, 'generate_quaternions': True } } """ Define each data configuration settings with a constraint (i.e. when setting A is X, setting B can't be Y) This will inform checks in various parts of the code path including both python and javascript. Format: for example when setting A is X, setting B must only be one of Y or Z would be written as: 'A': { 'X': { 'B': ['Y', 'Z'] }, } """ SETTING_CONSTRAINTS = { 'processing_mode': { STATIC: { 'troposphere_model': [TROP_GMF, TROP_GPT2, TROP_VMF1] }, KINEMATIC: { 'troposphere_model': [TROP_GMF, TROP_GPT2, TROP_VMF1] }, KINEMATIC_FAST: { 'ocean_loading': [False], 'troposphere_model': [TROP_GPT2, TROP_VMF1, TROP_PROVIDED, TROP_OFF] } } }
[docs]class GIPSYXData(GIPSYData): pass
[docs] @classmethod def get_verbose_marker_type(cls, marker_type): return get_verbose(marker_type, cls.MARKER_TYPES)
[docs] @classmethod def normalize_marker_type(cls, marker_type): return normalize(marker_type, cls.MARKER_TYPES)
MARKER_TYPE_DESCRIPTIONS = { None: 'Undefined', GEODETIC: 'Earth-fixed, high precision monumentation', NON_GEODETIC: 'Earth-fixed, low precision monumentation', NON_PHYSICAL: 'Generated from network processing', SPACEBORNE: 'Orbiting space vehicle', AIRBORNE: 'Aircraft, balloon, etc', WATER_CRAFT: 'Mobile water craft', GROUND_CRAFT: 'Mobile terrestrial vehicle', FIXED_BUOY: '"Fixed" on water surface', FLOATING_BUOY: 'Floating on water surface', FLOATING_ICE: 'Floating ice sheet, etc', GLACIER: '"Fixed" on a glacier', BALLISTIC: 'Rockets, shells, etc', ANIMAL: 'Animal carrying a receiver', HUMAN: 'Human being' } MARKER_TYPE_PROCESSING_MAPPING = { None: None, GEODETIC: GIPSYData.STATIC, NON_GEODETIC: GIPSYData.KINEMATIC_SLOW, NON_PHYSICAL: None, SPACEBORNE: GIPSYData.KINEMATIC_FAST, AIRBORNE: GIPSYData.KINEMATIC_FAST, WATER_CRAFT: GIPSYData.KINEMATIC_SLOW, GROUND_CRAFT: GIPSYData.KINEMATIC_SLOW, FIXED_BUOY: GIPSYData.STATIC, FLOATING_BUOY: GIPSYData.KINEMATIC_SLOW, FLOATING_ICE: GIPSYData.KINEMATIC_SLOW, GLACIER: GIPSYData.STATIC, BALLISTIC: GIPSYData.KINEMATIC_FAST, ANIMAL: GIPSYData.KINEMATIC_SLOW, HUMAN: GIPSYData.KINEMATIC_SLOW } GPS_TIME = 'G' GLONASS_TIME = 'R' GALILEO_TIME = 'E' QZSS_TIME = 'J' BEIDOU_TIME = 'C' IRNSS_TIME = 'I' TIME_STANDARDS = ( (GPS_TIME, 'GPS'), (GLONASS_TIME, 'GLO'), (GALILEO_TIME, 'GAL'), (QZSS_TIME, 'QZS'), (IRNSS_TIME, 'IRNSS'), (BEIDOU_TIME, 'BDT') )
[docs] @classmethod def get_verbose_time_standard(cls, time_standard): return get_verbose(time_standard, cls.TIME_STANDARDS)
[docs] @classmethod def normalize_time_standard(cls, time_standard): return normalize(time_standard, cls.TIME_STANDARDS)
GPS = 'G' GLONASS = 'R' GALILEO = 'E' QZSS = 'J' BEIDOU = 'C' IRNSS = 'I' SBAS = 'S' MIXED = 'M' GNSS_TYPES = ( (GPS, 'GPS'), (GLONASS, 'GLONASS'), (GALILEO, 'Galileo'), (QZSS, 'QZSS'), (BEIDOU, 'Beidou'), (IRNSS, 'IRNSS'), (SBAS, 'SBAS'), (MIXED, 'Mixed') )
[docs] @classmethod def get_verbose_gnss_types(cls, gnss_types): return get_verbose(gnss_types, cls.GNSS_TYPES)
[docs] @classmethod def normalize_gnss_types(cls, gnss_types): return normalize(gnss_types, cls.GNSS_TYPES)
[docs]class UserAlert(object): SUCCESS = 'S' INFO = 'I' WARN = 'W' ALARM = 'A' LEVELS = ( (SUCCESS, 'Success'), (INFO, 'Info'), (WARN, 'Warning'), (ALARM, 'Alarm'), )
[docs] @classmethod def get_verbose_level(cls, level): return get_verbose(level, cls.LEVELS)
[docs] @classmethod def normalize_level(cls, level): return normalize(level, cls.LEVELS)