Skip to content

Add type annotations #37

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
namurphy wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

Add type annotations #37

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1c7031c
Remove ty:ignore comments
namurphy Sep 30, 2026
613efa9
Add type annotations via Claude
namurphy Sep 30, 2026
34aa93f
Add ty:ignore comments
namurphy Sep 30, 2026
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
2 changes: 1 addition & 1 deletion noxfile.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
87 changes: 62 additions & 25 deletions src/pyfaradaycup/pipeline/ccsds_reader_pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
# $LastChangedBy: acase $
""" # ruff:ignore[D400]

from __future__ import annotations

__all__ = [
"apid_obj",
"choose_file",
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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]
Expand All @@ -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.
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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``.

Expand Down Expand Up @@ -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
)
Expand All @@ -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]
Expand All @@ -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.

Expand Down Expand Up @@ -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":
Expand Down
Loading
Loading