Skip to content

Commit

Permalink
fix: Docstrings + type hints in igslog.py
Browse files Browse the repository at this point in the history
  • Loading branch information
ronaldmaj committed Dec 23, 2024
1 parent 3ee4917 commit 62ce0dc
Showing 1 changed file with 15 additions and 9 deletions.
24 changes: 15 additions & 9 deletions gnssanalysis/gn_io/igslog.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import glob as _glob
import re as _re
from multiprocessing import Pool as _Pool
from typing import List as _List
from typing import Union, List, Tuple

import numpy as _np
import pandas as _pd
Expand Down Expand Up @@ -160,7 +160,7 @@ def determine_log_version(data: bytes) -> str:
raise LogVersionError("Log file does not conform to any known IGS version")


def extract_id_block(data: bytes, file_path: str, file_code: str, version: str = None) -> bytes:
def extract_id_block(data: bytes, file_path: str, file_code: str, version: str = None) -> Union[List[str], _np.array]:
"""Extract the site ID 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
Expand Down Expand Up @@ -193,7 +193,7 @@ def extract_id_block(data: bytes, file_path: str, file_code: str, version: str =
return id_block


def extract_location_block(data: bytes, file_path: str, version: str = None) -> bytes:
def extract_location_block(data: bytes, file_path: str, version: str = None) -> Union[_re.Match[bytes], _np.array]:
"""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
Expand All @@ -219,12 +219,12 @@ def extract_location_block(data: bytes, file_path: str, version: str = None) ->
return location_block


def extract_receiver_block(data: bytes, file_path: str) -> bytes:
def extract_receiver_block(data: bytes, file_path: str) -> Union[List[Tuple[bytes]], _np.array]:
"""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
:return bytes: The receiver block of the IGS site log
:return List[Tuple[bytes]]: The receiver block of the data. Each list element specifies an receiver
"""
receiver_block = _REGEX_REC.findall(data)
if receiver_block == []:
Expand All @@ -233,15 +233,21 @@ def extract_receiver_block(data: bytes, file_path: str) -> bytes:
return receiver_block


def extract_antenna_block(data: bytes, file_path: str) -> bytes:
def extract_antenna_block(data: bytes, file_path: str) -> Union[List[Tuple[bytes]], _np.array]:
"""Extract the antenna 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
:return List[Tuple[bytes]]: The antenna block of the data. Each list element specifies an antenna
"""
antenna_block = _REGEX_ANT.findall(data)
if antenna_block == []:
logger.warning(f"ANT rejected from {file_path}")
return _np.array([]).reshape(0, 12)
return antenna_block


def parse_igs_log(filename_array: _np.ndarray) -> _np.ndarray:
def parse_igs_log(filename_array: _np.ndarray) -> Union[_np.ndarray, None]:
"""Parses igs log and outputs ndarray with parsed data
:param _np.ndarray filename_array: Metadata on input log file. Expects ndarray of the form [CODE DATE PATH]
Expand Down Expand Up @@ -358,13 +364,13 @@ def translate_series(series: _pd.Series, translation: dict) -> _pd.Series:

def gather_metadata(
logs_glob_path: str = "/data/station_logs/station_logs_IGS/*/*.log", rnx_glob_path: str = None, num_threads: int = 1
) -> _List[_pd.DataFrame]:
) -> List[_pd.DataFrame]:
"""Parses log files found with glob expressions into pd.DataFrames
:param str logs_glob_path: A glob expression for log files, defaults to "/data/station_logs_IGS/*/*.log"
:param str rnx_glob_path: A glob expression for rnx files, e.g. /data/pea/exs/data/*.rnx, defaults to None
:param int num_threads: Number of threads to run, defaults to 1
:return _List[_pd.DataFrame]: List of DataFrames with [ID, Receiver, Antenna] data
:return List[_pd.DataFrame]: List of DataFrames with [ID, Receiver, Antenna] data
"""
parsed_filenames = find_recent_logs(logs_glob_path=logs_glob_path, rnx_glob_path=rnx_glob_path).values

Expand Down

0 comments on commit 62ce0dc

Please sign in to comment.