added apps clean, extract, header and moved some config to toml

pyproject.toml

@@ -10,3 +10,35 @@ build-backend = "setuptools.build_meta"
 [tool.black]
 line-length = 90
 target_version = ['py38', 'py39', 'py310']
+
+
+[tool.pytest.ini_options]
+minversion = "6.0"
+testpaths = [
+    "tests",
+]
+
+
+[tool.coverage.paths]
+source = ["./sigpyproc/"]
+
+[tool.coverage.run]
+omit = [
+    '*tests*',
+    '*docs*',
+    '*apps*',
+    '*setup.py',
+    '*__init__.py',
+    '*sigpyproc/core/kernels.py',
+]
+
+[tool.coverage.report]
+show_missing = true
+#fail_under = 85
+ignore_errors = true
+exclude_lines = ['raise AssertionError', 'raise NotImplementedError']
+
+
+[tool.mypy]
+ignore_missing_imports = true
+plugins = ["numpy.typing.mypy_plugin"]

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = sigpyproc
-version = 1.0.0
+version = 1.1.0
 author = Ewan Barr
 author_email = [email protected]
 maintainer = Pravir Kumar
@@ -31,6 +31,7 @@ install_requires =
     scipy
     bottleneck
     attrs
+    click
     rich
     bidict
     iqrm
@@ -40,6 +41,8 @@ install_requires =
 console_scripts =
     spp_header = sigpyproc.apps.spp_header:main
     spp_decimate = sigpyproc.apps.spp_decimate:main
+    spp_extract = sigpyproc.apps.spp_extract:main
+    spp_clean = sigpyproc.apps.spp_clean:main
 
 [options.extras_require]
 tests =
@@ -74,7 +77,7 @@ ignore =
     Q0,
     # WPS
     WPS100, WPS110, WPS114
-    WPS210, WPS220, WPS237,
+    WPS210, WPS216, WPS220, WPS237,
     WPS305, WPS323, WPS338, WPS339, WPS345, WPS352, WPS362
     WPS420, WPS432, WPS440, WPS441,
     WPS519,
@@ -114,31 +117,3 @@ rst-roles =
 docstring_style=numpy
 ignore=DAR402,DAR103,DAR201,DAR101
 
-
-[tool:pytest]
-minversion = 6
-testpaths = tests
-
-[coverage:run]
-omit =
-    *setup.py
-	*__init__.py
-	*tests*
-	*docs*
-    *apps*
-    *sigpyproc/core/kernels.py
-
-[coverage:report]
-show_missing = True
-ignore_errors = True
-#fail_under = 85
-exclude_lines =
-	raise AssertionError
-	raise NotImplementedError
-
-[coverage:paths]
-source = ./
-
-[mypy]
-ignore_missing_imports = True
-plugins = numpy.typing.mypy_plugin

sigpyproc/apps/spp_clean.py

@@ -0,0 +1,33 @@
+import click
+
+from sigpyproc.readers import FilReader
+
+
+@click.command(
+    context_settings=dict(help_option_names=["-h", "--help"], show_default=True)
+)
+@click.argument("filfile", type=click.Path(exists=True))
+@click.option(
+    "-m",
+    "--method",
+    type=click.Choice(["mad", "iqrm"]),
+    default="mad",
+    help="RFI cleaning method to use.",
+)
+@click.option(
+    "-t", "--threshold", type=float, default=3.0, help="Sigma threshold for RFI cleaning."
+)
+@click.option(
+    "-g", "--gulp", type=int, default=16384, help="Number of samples to read at once"
+)
+@click.option(
+    "-o", "--outfile", type=click.Path(exists=False), default=None, help="Output filename"
+)
+def main(filfile: str, method: str, threshold: float, outfile: str, gulp: int) -> None:
+    """Clean RFI from filterbank data."""
+    fil = FilReader(filfile)
+    fil.clean_rfi(method=method, threshold=threshold, outfile=outfile, gulp=gulp)
+
+
+if __name__ == "__main__":
+    main()

sigpyproc/apps/spp_decimate.py

@@ -1,44 +1,28 @@
-import argparse
+import click
 
 from sigpyproc.readers import FilReader
 
 
-def decimate(filename, tfactor=1, ffactor=1, gulp=16384, outfile=None):
-    fil = FilReader(filename)
-    fil.downsample(tfactor=tfactor, ffactor=ffactor, gulp=gulp, outfile=outfile)
-
-
-def main():
-    description = "Reduce time and/or frequency resolution of filterbank data."
-    parser = argparse.ArgumentParser(
-        description=description, formatter_class=argparse.ArgumentDefaultsHelpFormatter
-    )
-    parser.add_argument("filename", type=str, help="Path of the filterbank data file")
-
-    parser.add_argument("-o", "--outfile", type=str, help="Output file name")
-
-    parser.add_argument(
-        "-f", "--nchan", type=str, default=1, help="Number of frequency channels to add"
-    )
-
-    parser.add_argument(
-        "-t", "--nsamp", type=str, default=1, help="Number of time samples to add"
-    )
-
-    parser.add_argument("--nbits", type=str, help="Output number of bits per sample")
-
-    parser.add_argument(
-        "-g", "--gulp", type=str, default=16384, help="Number of samples to read at once"
-    )
-    args = parser.parse_args()
-
-    decimate(
-        args.filename,
-        tfactor=args.samp,
-        ffactor=args.nchan,
-        gulp=args.gulp,
-        outfile=args.outfile,
-    )
+@click.command(
+    context_settings=dict(help_option_names=["-h", "--help"], show_default=True)
+)
+@click.argument("filfile", type=click.Path(exists=True))
+@click.option(
+    "-t", "--tfactor", type=int, default=1, help="Number of time samples to add"
+)
+@click.option(
+    "-c", "--ffactor", type=int, default=1, help="Number of frequency channels to add"
+)
+@click.option(
+    "-g", "--gulp", type=int, default=16384, help="Number of samples to read at once"
+)
+@click.option(
+    "-o", "--outfile", type=click.Path(exists=False), default=None, help="Output filename"
+)
+def main(filfile: str, tfactor: int, ffactor: int, gulp: int, outfile: str) -> None:
+    """Reduce time and/or frequency resolution of filterbank data."""
+    fil = FilReader(filfile)
+    fil.downsample(tfactor=tfactor, ffactor=ffactor, gulp=gulp, filename=outfile)
 
 
 if __name__ == "__main__":

sigpyproc/apps/spp_extract.py

@@ -0,0 +1,29 @@
+import click
+
+from sigpyproc.readers import FilReader
+
+
+@click.command(
+    context_settings=dict(help_option_names=["-h", "--help"], show_default=True)
+)
+@click.argument("filfile", type=click.Path(exists=True))
+@click.option(
+    "-s", "--start", type=int, default=0, help="Start time sample"
+)
+@click.option(
+    "-n", "--nsamps", type=int, help="Number of time samples to extract"
+)
+@click.option(
+    "-g", "--gulp", type=int, default=16384, help="Number of samples to read at once"
+)
+@click.option(
+    "-o", "--outfile", type=click.Path(exists=False), default=None, help="Output filename"
+)
+def main(filfile: str, start: int, nsamps: int, gulp: int, outfile: str) -> None:
+    """Extract time samples from filterbank data."""
+    fil = FilReader(filfile)
+    fil.extract_samps(start=start, nsamps=nsamps, outfile=outfile, gulp=gulp)
+
+
+if __name__ == "__main__":
+    main()

sigpyproc/apps/spp_header.py

@@ -1,102 +1,38 @@
-import argparse
-
-from astropy import units
-from typing import Tuple
+from __future__ import annotations
+import click
 
 from sigpyproc.header import Header
-
-
-def get_duration_print(duration: float) -> Tuple[str, float]:
-    print_unit = "seconds"
-    if duration > 60:
-        duration /= 60
-        print_unit = "minutes"
-        if duration > 60:
-            duration /= 60
-            print_unit = "hours"
-            if duration > 24:
-                duration /= 24
-                print_unit = "days"
-    return print_unit, duration
-
-
-def header_string(header: Header) -> str:
-    print_hdr = []
-    temp = "{0:<33}: {1}"
-    print_hdr.extend(
-        [
-            temp.format("Data file", header.filename),
-            temp.format("Header size (bytes)", header.hdrlens[0]),
-            temp.format("Data size (bytes)", header.datalens[0]),
-            temp.format("Data type", f"{header.data_type} ({header.frame})"),
-            temp.format("Telescope", header.telescope),
-            temp.format("Datataking Machine", header.backend),
-            temp.format("Source Name", header.source),
-            temp.format("Source RA (J2000)", header.ra),
-            temp.format("Source DEC (J2000)", header.dec),
-            temp.format("Start AZ (deg)", header.azimuth.deg),
-            temp.format("Start ZA (deg)", header.zenith.deg),
-        ]
-    )
-
-    if header.data_type == "filterbank":
-        print_hdr.extend(
-            [
-                temp.format("Frequency of channel 1 (MHz)", header.fch1),
-                temp.format("Channel bandwidth      (MHz)", header.foff),
-                temp.format("Number of channels", header.nchans),
-                temp.format("Number of beams", header.nbeams),
-                temp.format("Beam number", header.ibeam),
-            ]
-        )
-    elif header.data_type == "time series":
-        print_hdr.extend(
-            [
-                temp.format("Reference DM (pc/cc)", header.dm),
-                temp.format("Reference frequency    (MHz)", header.fch1),
-                temp.format("Number of channels", header.nchans),
-            ]
-        )
-
-    print_unit, print_dur = get_duration_print(header.tobs)
-    print_hdr.extend(
-        [
-            temp.format("Time stamp of first sample (MJD)", header.tstart),
-            temp.format("Gregorian date (YYYY-MM-DD)", header.obs_date),
-            temp.format(
-                "Sample time (us)",
-                (header.tsamp * units.second).to(units.microsecond).value,
-            ),
-            temp.format("Number of samples", header.nsamples),
-            temp.format(f"Observation length {print_unit}", f"{print_dur:.1f}"),
-            temp.format("Number of bits per sample", header.nbits),
-            temp.format("Number of IFs", header.nifs),
-        ]
-    )
-    return "\n".join(print_hdr)
-
-
-def main():
-    description = "Examine header parameters of filterbank data."
-    parser = argparse.ArgumentParser(
-        description=description, formatter_class=argparse.ArgumentDefaultsHelpFormatter
-    )
-    parser.add_argument("filename", type=str, help="Path of the filterbank data file")
-
-    parser.add_argument(
-        "-k",
-        "--key",
-        type=str,
-        help="A header key to read (e.g. telescope, fch1, nsamples)",
-    )
-
-    args = parser.parse_args()
-
-    header = Header.from_sigproc(args.filename)
-    if args.key:
-        print(f"{args.key} = {getattr(header, args.key)}", flush=True)  # noqa: WPS421
-    else:
-        print(header_string(header), flush=True)  # noqa: WPS421
+from sigpyproc.io.sigproc import edit_header
+
+
+@click.group(context_settings=dict(help_option_names=["-h", "--help"], show_default=True))
+@click.argument("filfile", type=click.Path(exists=True))
+def main(filfile: str) -> None:
+    """Print or edit the header of a filterbank file."""
+    header = Header.from_sigproc(filfile)
+    click.echo(header.to_string())
+
+
+@main.command()
+@click.option(
+    "-k", "--key", type=str, help="A header key to read (e.g. telescope, fch1, nsamples)"
+)
+def get(filfile: str, key: str) -> None:
+    header = Header.from_sigproc(filfile)
+    click.echo(f"{key} = {getattr(header, key)}")
+
+
+@main.command()
+@click.option(
+    "-i",
+    "--item",
+    nargs=2,
+    type=click.Tuple([str, str]),
+    help="(key, value) to update in header",
+)
+def update(filfile: str, item: tuple[str, str]) -> None:
+    key, value = item
+    edit_header(filfile, key, value)
 
 
 if __name__ == "__main__":