diff --git a/gnssanalysis/gn_io/igslog.py b/gnssanalysis/gn_io/igslog.py index 7fe741c..8d5b2e7 100644 --- a/gnssanalysis/gn_io/igslog.py +++ b/gnssanalysis/gn_io/igslog.py @@ -5,6 +5,7 @@ import re as _re from multiprocessing import Pool as _Pool from typing import Optional, Union +import warnings import numpy as _np import pandas as _pd @@ -59,6 +60,7 @@ _re.IGNORECASE | _re.VERBOSE, ) +# See example here: https://files.igs.org/pub/station/general/blank.log _REGEX_LOC_V2 = _re.compile( rb""" 2.+\W+City\sor\sTown\s+\:\s*(\w[^\(\n\,/\?]+|).*\W+ @@ -222,15 +224,24 @@ def extract_id_block( return id_block -def extract_location_block(data: bytes, file_path: str, version: Union[str, None] = None) -> _np.ndarray: +def extract_location_block( + data: bytes, + file_path: str, + version: Union[str, None] = None, + raise_on_extract_failure: bool = True, + raise_on_unexpected_element_count: bool = True, +) -> list[str]: """Extract the location block given the bytes object read from an IGS site log file :param bytes data: The bytes object returned from an open() call on a IGS site log in "rb" mode :param str file_path: The path to the file from which the "data" bytes object was obtained :param str version: Version number of log file (e.g. "v2.0") - will be determined from input data unless provided here. + :param bool raise_on_extract_failure: raise (default) rather than just warning, if regex extract fails + :param bool raise_on_unexpected_element_count: raise (default) rather than just warning, if LOC elements count != 8 :raises LogVersionError: Raises an error if an unknown version string is passed in - :return _np.ndarray: The location block of the IGS site log, as a numpy NDArray of strings + :raises ValueError: If an the location block regex does not match, or if the number of elements extracted is not 8 + :return list[str]: The location block of the IGS site log, as a list of strings """ if version == None: version = determine_log_version(data) @@ -242,11 +253,26 @@ def extract_location_block(data: bytes, file_path: str, version: Union[str, None else: raise LogVersionError(f"Incorrect version string '{version}' passed to extract_location_block() function") - location_block = _REGEX_LOC.search(data) - if location_block is None: - logger.warning(f"LOC rejected from {file_path}") - return _np.array([]).reshape(0, 12) - return location_block + loc_block_match = _REGEX_LOC.search(data) + if loc_block_match is None: + if raise_on_extract_failure: + raise ValueError(f"Failed to extract LOC block from {file_path}") + warnings.warn(f"Failed to extract LOC block from {file_path}") + return [] + + # List of location properties. + # See example under '2. Site Location Information' here: https://files.igs.org/pub/station/general/blank.log + loc_block_decoded = [group.decode(encoding="utf8", errors="ignore") for group in loc_block_match.groups()] + + loc_element_count = len(loc_block_decoded) + if loc_element_count != 8: + # TODO consider using t-strings in Python 3.14 + loc_count_warning = "Expected 8 elements in LOC block, got: {block_length}, file: {file_path}" + if raise_on_unexpected_element_count: + raise ValueError(loc_count_warning.format(block_length=loc_element_count, file_path=file_path)) + warnings.warn(loc_count_warning.format(block_length=loc_element_count, file_path=file_path)) + + return loc_block_decoded def extract_receiver_block(data: bytes, file_path: str) -> Union[list[tuple[bytes]], _np.ndarray]: @@ -304,7 +330,6 @@ def parse_igs_log_data(data: bytes, file_path: str, file_code: str) -> Union[_np file_path=file_path, version=version, ) - blk_loc = [group.decode(encoding="utf8", errors="ignore") for group in blk_loc.groups()] # Combine ID and Location information: blk_id_loc = _np.asarray([0] + blk_id + blk_loc, dtype=object)[_np.newaxis] # Extract and re-format information from receiver block: diff --git a/tests/test_igslog.py b/tests/test_igslog.py index 5b8e5da..eba08c6 100644 --- a/tests/test_igslog.py +++ b/tests/test_igslog.py @@ -42,16 +42,51 @@ def test_extract_id_block(self): def test_extract_location_block(self): # Version 1 Location description results: v1_location_block = igslog.extract_location_block(v1_data, "/example/path", "v1.0") - self.assertEqual(v1_location_block.group(1), b"Les Abymes") - self.assertEqual(v1_location_block.group(2), b"Guadeloupe") + self.assertEqual(v1_location_block[0], "Les Abymes") + self.assertEqual(v1_location_block[1], "Guadeloupe") + + self.assertEqual(v1_location_block[2], "2919786.0") # X + self.assertEqual(v1_location_block[3], "-5383745.0") # Y + self.assertEqual(v1_location_block[4], "1774604.0") # Z + + self.assertEqual(len(v1_location_block), 8) + + # Source values: + # City or Town : Les Abymes + # State or Province : Guadeloupe (971) + # Country : Guadeloupe + # Tectonic Plate : CARIBBEAN + # Approximate Position (ITRF) + # X coordinate (m) : 2919786.0 + # Y coordinate (m) : -5383745.0 + # Z coordinate (m) : 1774604.0 + # Latitude (N is +) : +161544.30 + # Longitude (E is +) : -0613139.11 + # Elevation (m,ellips.) : -25.0 + + # V2: + # City or Town : Les Abymes + # State or Province : Guadeloupe (971) + # Country or Region : GLP + # Tectonic Plate : CARIBBEAN + # Approximate Position (ITRF) + # X coordinate (m) : 2919786.0 + # Y coordinate (m) : -5383745.0 + # Z coordinate (m) : 1774604.0 + # Latitude (N is +) : +161544.30 + # Longitude (E is +) : -0613139.11 + # Elevation (m,ellips.) : -25.0 + # Additional Information : # Version 2 Location description results: v2_location_block = igslog.extract_location_block(v2_data, "/example/path", "v2.0") - self.assertEqual(v2_location_block.group(1), b"Les Abymes") - self.assertEqual(v2_location_block.group(2), b"GLP") + self.assertEqual(v2_location_block[0], "Les Abymes") + self.assertEqual(v2_location_block[1], "GLP") - # Coordinate information remains the same: - self.assertEqual(v2_location_block.group(3), v1_location_block.group(3)) + # Coordinate information remains the same compared to v1.0: + self.assertEqual(v2_location_block[2], v1_location_block[2]) + self.assertEqual(v2_location_block[3], v1_location_block[3]) + self.assertEqual(v2_location_block[4], v1_location_block[4]) # Check LogVersionError is rasied on no data: with self.assertRaises(igslog.LogVersionError):