Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 17 additions & 11 deletions src/pyfaradaycup/decommutator/ccsds_reader_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,8 @@

def _file2bytestr(
path: str | os.PathLike[str],
gzip: bool = False, # ruff:ignore[FBT001, FBT002]
*,
gzip: bool = False,
) -> bytes:
"""
Read the entire contents of a file into a bytes object.
Expand Down Expand Up @@ -104,8 +105,9 @@ def _choose_file(path: str | os.PathLike[str]) -> str:

def _wrapper_status(
path: str | os.PathLike[str],
gzip: bool = False, # ruff:ignore[FBT001, FBT002]
spconly: bool = False, # ruff:ignore[FBT001, FBT002]
*,
gzip: bool = False,
spconly: bool = False,
) -> dict[str, list[int]]:
"""
Read CCSDS headers from SWEM wrapper packets and the packets inside them.
Expand Down Expand Up @@ -198,8 +200,9 @@ def _wrapper_status(

def read_file( # ruff:ignore[C901]
path: str | os.PathLike[str],
verbose: bool = False, # ruff:ignore[FBT001, FBT002]
gzip: bool = False, # ruff:ignore[FBT001, FBT002]
*,
verbose: bool = False,
gzip: bool = False,
) -> dict[int, dict[str, list]]:
"""
Read SPC instrument packets from an L0 file.
Expand Down Expand Up @@ -326,9 +329,10 @@ def read_file( # ruff:ignore[C901]

def read_file_sc( # ruff:ignore[C901, PLR0912, PLR0915]
path: str | os.PathLike[str],
verbose: bool = False, # ruff:ignore[FBT001, FBT002]
ptp: bool = False, # ruff:ignore[FBT001, FBT002]
gzip: bool = False, # ruff:ignore[FBT001, FBT002]
*,
verbose: bool = False,
ptp: bool = False,
gzip: bool = False,
) -> dict[int, dict[str, list]]:
"""
Read spacecraft housekeeping packets from an L0 file.
Expand Down Expand Up @@ -575,7 +579,8 @@ def _read_bytestr( # ruff:ignore[C901, PLR0912, PLR0913]
data: dict[int, dict[str, list]],
apidformat, # ruff:ignore[ANN001]
pktcnt: list[dict[int, int]],
verbose: bool = False, # ruff:ignore[FBT001, FBT002]
*,
verbose: bool = False,
) -> tuple[()]:
"""Take a hex string and find packets""" # ruff:ignore[D400]
# Parse the CCSDS header
Expand Down Expand Up @@ -853,7 +858,7 @@ def __init__(self) -> None:
self.startbit: list[int] = []


def _get_layout(apid: int, verbose: bool = False) -> apid_obj | None: # ruff:ignore[C901, FBT001, FBT002]
def _get_layout(apid: int, *, verbose: bool = False) -> apid_obj | None: # ruff:ignore[C901]
"""
Read the bit layout for one SWEAP APID from ``sweap_tlm.blk``.

Expand Down Expand Up @@ -942,7 +947,8 @@ def _get_layout(apid: int, verbose: bool = False) -> apid_obj | None: # ruff:ig

def _get_layout_sc( # ruff:ignore[C901]
apid: int,
verbose: bool = False, # ruff:ignore[FBT001, FBT002]
*,
verbose: bool = False,
filename: str | os.PathLike[str] | None = None,
) -> tuple[apid_obj, int] | None:
"""
Expand Down
40 changes: 23 additions & 17 deletions src/pyfaradaycup/decommutator/swp_spc_l0_to_l05.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,16 +61,17 @@
logfile: TextIO


def main( # ruff:ignore[C901, PLR0912, PLR0913, PLR0915, PLR0917]
def main( # ruff:ignore[C901, PLR0912, PLR0913, PLR0915]
l0file: str | os.PathLike[str],
l05dir: str | os.PathLike[str] | None = None,
logdir: str | os.PathLike[str] | None = None,
spacecraft: bool = False, # ruff:ignore[FBT001, FBT002]
ptp: bool = False, # ruff:ignore[FBT001, FBT002]
gzip: bool = False, # ruff:ignore[FBT001, FBT002]
*,
spacecraft: bool = False,
ptp: bool = False,
gzip: bool = False,
apidreq: int = 0,
overwrite: bool = False, # ruff:ignore[FBT001, FBT002]
verbose: bool = False, # ruff:ignore[FBT001, FBT002]
overwrite: bool = False,
verbose: bool = False,
) -> None:
"""
Convert one SPC L0 file into L0.5 CDF files, one per APID.
Expand Down Expand Up @@ -367,7 +368,7 @@ def main( # ruff:ignore[C901, PLR0912, PLR0913, PLR0915, PLR0917]
logfile.close()


def _cdf35e_35f(cdf: pycdf.CDF, dat: dict[str, list], verbose: bool = False) -> None: # ruff:ignore[ARG001, C901, FBT001, FBT002]
def _cdf35e_35f(cdf: pycdf.CDF, dat: dict[str, list], *, verbose: bool = False) -> None: # ruff:ignore[ARG001, C901]
"""
Fill a CDF with housekeeping data, one row per packet.

Expand Down Expand Up @@ -466,8 +467,9 @@ def _cdf35e_35f(cdf: pycdf.CDF, dat: dict[str, list], verbose: bool = False) ->
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]
*,
nocdf: bool = False,
verbose: bool = False,
) -> dict[str, list] | None:
"""
Expand SPC science packets (APIDs 0x351, 0x353, 0x354) and write them to a CDF.
Expand Down Expand Up @@ -668,8 +670,9 @@ def _cdf351_353_354( # ruff:ignore[C901, PLR0912, PLR0915, RET503]
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]
*,
nocdf: bool = False,
verbose: bool = False,
) -> dict[str, list] | tuple[()]:
"""
Expand SPC time series (APID 0x352) packets into L0.5 data and write them to a CDF.
Expand Down Expand Up @@ -866,7 +869,8 @@ def _cdf352( # ruff:ignore[C901, PLR0912, PLR0915]
def _secsubsec2scet(
sec: list[int],
subsec: list[int],
spacecraft: bool = False, # ruff:ignore[FBT001, FBT002]
*,
spacecraft: bool = False,
) -> list[float]:
"""
Convert MET seconds and subseconds to ephemeris time in nanoseconds.
Expand Down Expand Up @@ -920,9 +924,10 @@ def _secsubsec2scet(

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]
*,
screen: bool = False,
file: bool = True,
verbose: bool = False,
) -> None:
"""
Write a timestamped status message to the log file and/or the screen.
Expand Down Expand Up @@ -958,8 +963,9 @@ def _statusmsg(


def _get_newest_kernel(
tls: bool = False, # ruff:ignore[FBT001, FBT002]
sclk: bool = False, # ruff:ignore[FBT001, FBT002]
*,
tls: bool = False,
sclk: bool = False,
) -> str:
"""
Find the newest NAIF leap second or PSP clock (SCLK) kernel file.
Expand Down
Loading