diff --git a/noxfile.py b/noxfile.py index 234d783..e83ab39 100644 --- a/noxfile.py +++ b/noxfile.py @@ -61,7 +61,7 @@ def tests(session: nox.Session) -> None: if RUNNING_ON_RTD: - rtd_output_path = pathlib.Path(os.environ.get("READTHEDOCS_OUTPUT")) / "html" # ty:ignore[invalid-argument-type] + rtd_output_path = pathlib.Path(os.environ.get("READTHEDOCS_OUTPUT")) / "html" # ty: ignore[invalid-argument-type] rtd_output_path.mkdir(parents=True, exist_ok=True) doc_build_dir = str(rtd_output_path) else: diff --git a/src/pyfaradaycup/pipeline/ccsds_reader_pipeline.py b/src/pyfaradaycup/pipeline/ccsds_reader_pipeline.py index 59e213b..9fdc54e 100644 --- a/src/pyfaradaycup/pipeline/ccsds_reader_pipeline.py +++ b/src/pyfaradaycup/pipeline/ccsds_reader_pipeline.py @@ -5,6 +5,8 @@ # $LastChangedBy: acase $ """ # ruff:ignore[D400] +from __future__ import annotations + __all__ = [ "apid_obj", "choose_file", @@ -38,11 +40,11 @@ # import tkFileDialog -def read_stdin(ptp=False, verbose=False): # ruff:ignore[ANN001, ANN201, FBT002] +def read_stdin(ptp: bool = False, verbose: bool = False) -> None: # ruff:ignore[FBT001, FBT002] """Parse binary stream on stdin""" # ruff:ignore[D400] -def file2bytestr(path="", verbose=False, gzip=False): # ruff:ignore[ANN001, ANN201, ARG001, FBT002] +def file2bytestr(path: str = "", verbose: bool = False, gzip: bool = False) -> bytes: # ruff:ignore[ARG001, FBT001, FBT002] """ Read the entire contents of a file into a bytes object. @@ -70,9 +72,9 @@ def file2bytestr(path="", verbose=False, gzip=False): # ruff:ignore[ANN001, ANN """ try: if gzip: - import gzip # ruff:ignore[PLC0415] + import gzip # ruff:ignore[PLC0415] # ty: ignore[invalid-assignment] - with gzip.open(path, "rb") as f: + with gzip.open(path, "rb") as f: # ty: ignore[unresolved-attribute] bytestr = f.read() return bytestr # ruff:ignore[RET504] with open(path, "rb") as f: # ruff:ignore[PTH123] @@ -85,7 +87,7 @@ def file2bytestr(path="", verbose=False, gzip=False): # ruff:ignore[ANN001, ANN sys.exit() -def choose_file(path="", ptp=False, verbose=False): # ruff:ignore[ANN001, ANN201, ARG001, FBT002] +def choose_file(path: str = "", ptp: bool = False, verbose: bool = False) -> str: # ruff:ignore[ARG001, FBT001, FBT002] # make sure file exists """ Check that a file can be opened and return its path. @@ -130,7 +132,12 @@ def choose_file(path="", ptp=False, verbose=False): # ruff:ignore[ANN001, ANN20 return path -def wrapper_status(path="", verbose=False, gzip=False, spconly=False): # ruff:ignore[ANN001, ANN201, ARG001, FBT002] +def wrapper_status( + path: str = "", + verbose: bool = False, # ruff:ignore[ARG001, FBT001, FBT002] + gzip: bool = False, # ruff:ignore[FBT001, FBT002] + spconly: bool = False, # ruff:ignore[FBT001, FBT002] +) -> dict[str, list[int]]: """ Read CCSDS headers from SWEM wrapper packets and the packets inside them. @@ -226,7 +233,11 @@ def wrapper_status(path="", verbose=False, gzip=False, spconly=False): # ruff:i return data -def read_file(path="", verbose=False, gzip=False): # ruff:ignore[ANN001, ANN201, C901, FBT002] +def read_file( # ruff:ignore[C901] + path: str = "", + verbose: bool = False, # ruff:ignore[FBT001, FBT002] + gzip: bool = False, # ruff:ignore[FBT001, FBT002] +) -> dict[int, dict[str, list]]: """Read a CCSDS File and return data structure""" # ruff:ignore[D400] # get a filename if not specified path = choose_file(path) @@ -247,7 +258,7 @@ def read_file(path="", verbose=False, gzip=False): # ruff:ignore[ANN001, ANN201 apidformat[apid] = get_layout(apid, verbose=verbose) if apidformat[apid]: data[apid] = {} - for name in apidformat[apid].names: + for name in apidformat[apid].names: # ty: ignore[unresolved-attribute] data[apid][name] = [] # create a list of two dictionaries that can keep track of @@ -317,7 +328,12 @@ def read_file(path="", verbose=False, gzip=False): # ruff:ignore[ANN001, ANN201 return data -def read_file_sc(path="", verbose=False, ptp=False, gzip=False): # ruff:ignore[ANN001, ANN201, C901, FBT002, PLR0912, PLR0915] +def read_file_sc( # ruff:ignore[C901, PLR0912, PLR0915] + path: str = "", + verbose: bool = False, # ruff:ignore[FBT001, FBT002] + ptp: bool = False, # ruff:ignore[FBT001, FBT002] + gzip: bool = False, # ruff:ignore[FBT001, FBT002] +) -> dict[int, dict[str, list]]: """Read a CCSDS File and return data structure""" # ruff:ignore[D400] # get a filename if not specified path = choose_file(path) @@ -409,7 +425,7 @@ def read_file_sc(path="", verbose=False, ptp=False, gzip=False): # ruff:ignore[ apid, verbose=verbose, filename=os.path.join("sc_hk_def", sc_hk_filename), # ruff:ignore[PTH118] - ) + ) # ty: ignore[not-iterable] if apidformat[apid]: data[apid] = {} for name in apidformat[apid].names: @@ -511,7 +527,14 @@ def read_file_sc(path="", verbose=False, ptp=False, gzip=False): # ruff:ignore[ return data -def read_bytestr(bytestr, pointer, data, apidformat, pktcnt, verbose=False): # ruff:ignore[ANN001, ANN201, C901, FBT002, PLR0912, PLR0913] +def read_bytestr( # ruff:ignore[C901, PLR0912, PLR0913] + bytestr: bytes, + pointer, # ruff:ignore[ANN001] + data: dict[int, dict[str, list]], + apidformat, # ruff:ignore[ANN001] + pktcnt: list[dict[int, int]], + verbose: bool = False, # ruff:ignore[FBT001, FBT002] +) -> tuple[()]: """Take a hex string and find packets""" # ruff:ignore[D400] # Parse the CCSDS header try: @@ -566,7 +589,7 @@ def read_bytestr(bytestr, pointer, data, apidformat, pktcnt, verbose=False): # return () -def parse_ccsds_head(bytestr, verbose=False): # ruff:ignore[ANN001, ANN201, ARG001, FBT002] +def parse_ccsds_head(bytestr: bytes, verbose: bool = False) -> dict[str, int]: # ruff:ignore[ARG001, FBT001, FBT002] """ Decode a 10-byte CCSDS packet header into its fields. @@ -619,7 +642,14 @@ def parse_ccsds_head(bytestr, verbose=False): # ruff:ignore[ANN001, ANN201, ARG return head -def parse_pkt(bytestr, data, apidformat, apid, ccsds_head, verbose=False): # ruff:ignore[ANN001, ANN201, ARG001, C901, FBT002, PLR0912, PLR0913] +def parse_pkt( # ruff:ignore[C901, PLR0912, PLR0913] + bytestr: bytes, + data: dict[int, dict[str, list]], + apidformat, # ruff:ignore[ANN001] + apid: int, + ccsds_head: dict[str, int], # ruff:ignore[ARG001] + verbose: bool = False, # ruff:ignore[ARG001, FBT001, FBT002] +) -> None: """Parse one CCSDS packet""" # ruff:ignore[D400] # The format for this APIDs packet list form = apidformat[apid] @@ -757,19 +787,22 @@ class apid_obj: # ruff:ignore[N801] packets that have one. """ - def __init__(self): # ruff:ignore[ANN204] - self.names = [] - self.bits = [] + apid: int + sw_data_vars: list[str] + + def __init__(self) -> None: + self.names: list[str] = [] + self.bits: list[int] = [] self.bytestart = [] self.bitstart = [] self.byteend = [] self.bitend = [] - self.data = {} - self.startbyte = [] - self.startbit = [] + self.data: dict[str, list] = {} + self.startbyte: list[int] = [] + self.startbit: list[int] = [] -def get_layout(apid, verbose=False): # ruff:ignore[ANN001, ANN201, C901, FBT002] +def get_layout(apid: int, verbose: bool = False) -> apid_obj | None: # ruff:ignore[C901, FBT001, FBT002] """ Read the bit layout for one SWEAP APID from ``sweap_tlm.blk``. @@ -818,7 +851,7 @@ def get_layout(apid, verbose=False): # ruff:ignore[ANN001, ANN201, C901, FBT002 if verbose: print(f"APID {hex(apid)[2:]} Format Found".upper()) # ruff:ignore[FURB116, T201] thisapid = apid_obj() - thisapid.apid = apid # ty: ignore[unresolved-attribute] + thisapid.apid = apid line = ( # ruff:ignore[PLW2901] "" # so that the while loop will start out ok ) @@ -832,11 +865,11 @@ def get_layout(apid, verbose=False): # ruff:ignore[ANN001, ANN201, C901, FBT002 thisapid.bits.append(int(pieces[3].strip())) thisapid.data[pieces[0].strip()] = [] if hasattr(thisapid, "sw_data_vars"): - thisapid.sw_data_vars.append(thisapid.names[-1]) # ty: ignore[unresolved-attribute] + thisapid.sw_data_vars.append(thisapid.names[-1]) elif (line.strip()[0:9] == "( SW_DATA") | ( line.strip()[0:12] == "( SW_SPC_SCI" ): - thisapid.sw_data_vars = [] # ty: ignore[unresolved-attribute] + thisapid.sw_data_vars = [] except IndexError: break except: # ruff:ignore[E722] @@ -862,7 +895,11 @@ def get_layout(apid, verbose=False): # ruff:ignore[ANN001, ANN201, C901, FBT002 return None -def get_layout_sc(apid, verbose=False, filename=""): # ruff:ignore[ANN001, ANN201, C901, FBT002] +def get_layout_sc( # ruff:ignore[C901] + apid: int, + verbose: bool = False, # ruff:ignore[FBT001, FBT002] + filename: str = "", +) -> tuple[apid_obj, int] | None: """ Read the bit layout for one spacecraft housekeeping APID. @@ -906,7 +943,7 @@ def get_layout_sc(apid, verbose=False, filename=""): # ruff:ignore[ANN001, ANN2 if verbose: print(f"APID {hex(apid)[2:]} Format Found".upper()) # ruff:ignore[FURB116, T201] thisapid = apid_obj() - thisapid.apid = apid # ty: ignore[unresolved-attribute] + thisapid.apid = apid line = "" # ruff:ignore[PLW2901] while line[0:4] != "SC_H": diff --git a/src/pyfaradaycup/pipeline/swp_spc_l02l1.py b/src/pyfaradaycup/pipeline/swp_spc_l02l1.py index 3672bf4..780d285 100644 --- a/src/pyfaradaycup/pipeline/swp_spc_l02l1.py +++ b/src/pyfaradaycup/pipeline/swp_spc_l02l1.py @@ -5,6 +5,8 @@ # $LastChangedBy: acase $ """ # ruff:ignore[D400] +from __future__ import annotations + __all__ = [ "cdf35e_35f", "cdf351_353_354", @@ -24,6 +26,7 @@ import os import pathlib import sys +from typing import TextIO import numpy as np import spiceypy @@ -33,6 +36,9 @@ DATA_DIR = pathlib.Path(__file__).parent.parent / "data" +# The log file, which is opened by main +logfile: TextIO + # Purpose: Convert binary "level-zero" or "ssr" files that come from the SWEM or Spacecraft # into L0.5 or L1 CDF files @@ -54,17 +60,17 @@ # - Added revision history -def main( # ruff:ignore[ANN201, C901, PLR0912, PLR0913, PLR0915, PLR0917] - l0file="", # ruff:ignore[ANN001] - l1dir="", # ruff:ignore[ANN001] - logdir="", # ruff:ignore[ANN001] - spacecraft=False, # ruff:ignore[ANN001, FBT002] - ptp=False, # ruff:ignore[ANN001, FBT002] - gzip=False, # ruff:ignore[ANN001, FBT002] - apidreq=0, # ruff:ignore[ANN001] - overwrite=False, # ruff:ignore[ANN001, FBT002] - verbose=False, # ruff:ignore[ANN001, FBT002] -): +def main( # ruff:ignore[C901, PLR0912, PLR0913, PLR0915, PLR0917] + l0file: str = "", + l1dir: str = "", + logdir: str = "", + spacecraft: bool = False, # ruff:ignore[FBT001, FBT002] + ptp: bool = False, # ruff:ignore[FBT001, FBT002] + gzip: bool = False, # ruff:ignore[FBT001, FBT002] + apidreq: int = 0, + overwrite: bool = False, # ruff:ignore[FBT001, FBT002] + verbose: bool = False, # ruff:ignore[FBT001, FBT002] +) -> None: """ Convert one SPC L0 file into L1 CDF files, one per APID. @@ -138,7 +144,7 @@ def main( # ruff:ignore[ANN201, C901, PLR0912, PLR0913, PLR0915, PLR0917] f"swp_spc_l02l1_{nowdt.year:04.0f}{nowdt.month:02.0f}{nowdt.day:02.0f}{nowdt.hour:02.0f}{nowdt.minute:02.0f}{nowdt.second:02.0f}.log", ) try: - global logfile # ruff:ignore[PLW0603] # ty:ignore[unresolved-global] + global logfile # ruff:ignore[PLW0603] logfile = open(logpath, "w") # ruff:ignore[PTH123, SIM115] except: # ruff:ignore[E722] print("\n***ERROR*** Could not open log file!\n") # ruff:ignore[T201] @@ -303,13 +309,13 @@ def main( # ruff:ignore[ANN201, C901, PLR0912, PLR0913, PLR0915, PLR0917] # Create a new CDF file from the provided skeleton try: cdf = pycdf.CDF(l1path, skeleton_filename) - except "CDFError": # ruff:ignore[B030] # ty:ignore[invalid-exception-caught] + except "CDFError": # ruff:ignore[B030] # ty: ignore[invalid-exception-caught] statusmsg( f"\n***ERROR*** [swp_spc_l02l1] Could not create new CDF (APID={apid})...continuing to next APID\n).", screen=True, verbose=verbose, ) - statusmsg(sys.exc_info(), screen=True, verbose=verbose) + statusmsg(sys.exc_info(), screen=True, verbose=verbose) # ty: ignore[invalid-argument-type] continue # Run a different procedure to put data into CDF file depending on APID @@ -349,7 +355,7 @@ def main( # ruff:ignore[ANN201, C901, PLR0912, PLR0913, PLR0915, PLR0917] logfile.close() -def cdf35e_35f(cdf, dat, verbose=False) -> None: # ruff:ignore[ANN001, C901, FBT002] +def cdf35e_35f(cdf: pycdf.CDF, dat: dict[str, list], verbose: bool = False) -> None: # ruff:ignore[C901, FBT001, FBT002] """ Fill a CDF with housekeeping data, one row per packet. @@ -446,7 +452,12 @@ def cdf35e_35f(cdf, dat, verbose=False) -> None: # ruff:ignore[ANN001, C901, FB statusmsg(sys.exc_info()) -def cdf351_353_354(cdf, dat, nocdf=False, verbose=False): # ruff:ignore[ANN001, ANN201, C901, FBT002, PLR0912, PLR0915, RET503] +def cdf351_353_354( # ruff:ignore[C901, PLR0912, PLR0915, RET503] + cdf: pycdf.CDF, + dat: dict[str, list], + nocdf: bool = False, # ruff:ignore[FBT001, FBT002] + verbose: bool = False, # ruff:ignore[FBT001, FBT002] +) -> dict[str, list] | None: """ Expand SPC science packets (APIDs 0x351, 0x353, 0x354) and write them to a CDF. @@ -641,7 +652,12 @@ def cdf351_353_354(cdf, dat, nocdf=False, verbose=False): # ruff:ignore[ANN001, raise RuntimeError # ruff:ignore[B904] -def cdf352(cdf, dat, nocdf=False, verbose=False): # ruff:ignore[ANN001, ANN201, C901, FBT002, PLR0912, PLR0915] +def cdf352( # ruff:ignore[C901, PLR0912, PLR0915] + cdf: pycdf.CDF, + dat: dict[str, list], + nocdf: bool = False, # ruff:ignore[FBT001, FBT002] + verbose: bool = False, # ruff:ignore[FBT001, FBT002] +) -> dict[str, list] | tuple[()]: """ Expand SPC time series (APID 0x352) packets into L1 data and write them to a CDF. @@ -819,7 +835,12 @@ def cdf352(cdf, dat, nocdf=False, verbose=False): # ruff:ignore[ANN001, ANN201, return () -def secsubsec2scet(sec, subsec, spacecraft=False, verbose=False): # ruff:ignore[ANN001, ANN201, ARG001, FBT002] +def secsubsec2scet( + sec: list[int], + subsec: list[int], + spacecraft: bool = False, # ruff:ignore[FBT001, FBT002] + verbose: bool = False, # ruff:ignore[ARG001, FBT001, FBT002] +) -> list[float]: """ Convert MET seconds and subseconds to ephemeris time in nanoseconds. @@ -873,7 +894,12 @@ def secsubsec2scet(sec, subsec, spacecraft=False, verbose=False): # ruff:ignore return ephem_nanosec_j2000 # ruff:ignore[RET504] -def statusmsg(string, screen=False, file=True, verbose=False): # ruff:ignore[ANN001, ANN201, FBT002] +def statusmsg( + string: str, + screen: bool = False, # ruff:ignore[FBT001, FBT002] + file: bool = True, # ruff:ignore[FBT001, FBT002] + verbose: bool = False, # ruff:ignore[FBT001, FBT002] +) -> None: """ Write a timestamped status message to the log file and/or the screen. @@ -907,7 +933,11 @@ def statusmsg(string, screen=False, file=True, verbose=False): # ruff:ignore[AN print(string) # ruff:ignore[T201] -def get_newest_kernel(tls=False, sclk=False, verbose=False): # ruff:ignore[ANN001, ANN201, ARG001, FBT002] +def get_newest_kernel( + tls: bool = False, # ruff:ignore[FBT001, FBT002] + sclk: bool = False, # ruff:ignore[FBT001, FBT002] + verbose: bool = False, # ruff:ignore[ARG001, FBT001, FBT002] +) -> str: """ Find the newest NAIF leap second or PSP clock (SCLK) kernel file. @@ -976,7 +1006,7 @@ def get_newest_kernel(tls=False, sclk=False, verbose=False): # ruff:ignore[ANN0 return path # ruff:ignore[RET504] -def get_newest_skeleton(apid, verbose=False): # ruff:ignore[ANN001, ANN201, ARG001, FBT002] +def get_newest_skeleton(apid: int, verbose: bool = False) -> str: # ruff:ignore[ARG001, FBT001, FBT002] """ Return the path to the skeleton CDF file for an APID. @@ -1030,7 +1060,7 @@ def get_newest_skeleton(apid, verbose=False): # ruff:ignore[ANN001, ANN201, ARG # return(path) -def setup(): # ruff:ignore[ANN201] +def setup() -> argparse.Namespace: """ Read the command-line arguments for running this module as a script. @@ -1170,13 +1200,13 @@ def setup(): # ruff:ignore[ANN201] statusmsg( "***ERROR*** You must provide --l0file, if not using -b or -r", screen=True, - verbose=verbose, # ruff:ignore[F821] # ty:ignore[unresolved-reference] + verbose=verbose, # ruff:ignoreignore[F821] # ty: ignore[unresolved-reference] ) elif args.l0dir == "": statusmsg( "***ERROR*** You must provide --l0dir if using -b or -r", screen=True, - verbose=verbose, # ruff:ignore[F821] # ty:ignore[unresolved-reference] + verbose=verbose, # ruff:ignore[F821] # ty: ignore[unresolved-reference] ) # Convert APID to an integer (it is read as a string from the command line) @@ -1190,9 +1220,9 @@ def setup(): # ruff:ignore[ANN201] statusmsg( "Trouble parsing desired APID....exiting.", screen=True, - verbose=verbose, # ruff:ignore[F821] # ty:ignore[unresolved-reference] + verbose=verbose, # ruff:ignore[F821] # ty: ignore[unresolved-reference] ) - statusmsg(sys.exc_info(), screen=True, verbose=verbose) # ruff:ignore[F821] # ty:ignore[unresolved-reference] + statusmsg(sys.exc_info(), screen=True, verbose=verbose) # ruff:ignore[F821] # ty: ignore[invalid-argument-type, unresolved-reference] sys.exit() # Make sure the environmental variable reference to the data directory is set and readable