#160 Merge pull request from deshima-dev/astropenguin/issue115

Add data package parser

CITATION.cff

@@ -31,5 +31,5 @@ keywords:
   - spectroscopy
   - deshima
 license: MIT
-version: 2024.7.1
-date-released: '2024-07-17'
+version: 2024.7.2
+date-released: '2024-07-22'

README.md

@@ -11,7 +11,7 @@ DESHIMA merge code for observed datasets
 ## Installation
 
 ```shell
-pip install demerge==2024.7.1
+pip install demerge==2024.7.2
 ```
 
 ## Command line interface

demerge/__init__.py

@@ -1,5 +1,5 @@
-__all__ = ["demerge", "merge", "reduce"]
-__version__ = "2024.7.1"
+__all__ = ["data", "demerge", "merge", "reduce"]
+__version__ = "2024.7.2"
 
 
 # standard library
@@ -8,12 +8,16 @@
 from logging import DEBUG, basicConfig, getLogger
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import Any, Literal, Optional
+from typing import Any, Literal, Optional, Union
 
 
 # dependencies
 from fire import Fire
-from . import merge, reduce
+from . import data, merge, reduce
+
+
+# type hints
+PathLike = Union[Path, str]
 
 
 # constants
@@ -22,7 +26,7 @@
 
 
 @contextmanager
-def set_dir(dir: Optional[Path] = None, /) -> Iterator[Path]:
+def set_dir(dir: Optional[PathLike] = None, /) -> Iterator[Path]:
     """Resolve a directory or set a temporary directory."""
     if dir is None:
         with TemporaryDirectory() as temp_dir:
@@ -50,10 +54,10 @@ def demerge(
     /,
     *,
     # data paths
-    data_dir: Path = Path(),
-    dems_dir: Path = Path(),
+    data_dir: PathLike = Path(),
+    dems_dir: PathLike = Path(),
     reduced_dir: Optional[Path] = None,
-    ddb: Path = PACKAGE_DATA / "ddb_20240713.fits.gz",
+    ddb: PathLike = PACKAGE_DATA / "ddb_20240713.fits.gz",
     # merge options
     measure: Literal["df/f", "brightness"] = "df/f",
     overwrite: bool = False,
@@ -64,11 +68,11 @@ def demerge(
 
     Args:
         obsid: Observation ID (YYYYmmddHHMMSS).
-        data_dir: Path where raw data directory is placed,
+        data_dir: Path of directory where data packages are placed,
             i.e. expecting ``${data_dir}/cosmos_YYYYmmddHHMMSS``.
-        dems_dir: Path where merged DEMS file will be placed,
+        dems_dir: Path of directory where merged DEMS will be placed,
             i.e. expecting ``${dems_dir}/dems_YYYYmmddHHMMSS.zarr.zip``.
-        reduced_dir: Path where reduced data directory will be placed,
+        reduced_dir: Path of directory where reduced packages are placed,
             i.e. expecting ``${reduced_dir}/reduced_YYYYmmddHHMMSS``.
             If not specified, a temporary directory will be used.
         ddb: Path of DDB (DESHIMA database) file.
@@ -91,59 +95,32 @@ def demerge(
         set_dir(dems_dir) as dems_dir,
         set_dir(reduced_dir) as reduced_dir,
     ):
-        data_dir_ = Path(data_dir).resolve() / f"cosmos_{obsid}"
-        dems_dir_ = Path(dems_dir).resolve()
-        reduced_dir_ = reduced_dir / f"reduced_{obsid}"
+        data_pack = Path(data_dir).resolve() / f"cosmos_{obsid}"
+        reduced_pack = reduced_dir / f"reduced_{obsid}"
+        data_pack_ = data.parse(data_pack)
 
         # Run reduce function
         readout = reduce.reduce(
-            data_dir=data_dir_,
-            reduced_dir=reduced_dir_,
+            data_pack=data_pack,
+            reduced_pack=reduced_pack,
             overwrite=overwrite,
             debug=debug,
         )
 
         # Run merge function
-        if (dems := dems_dir_ / f"dems_{obsid}.zarr.zip").exists() and not overwrite:
-            raise FileExistsError(dems)
-
-        if not (corresp := data_dir_ / "kid_corresp.json").exists():
-            raise FileNotFoundError(corresp)
-
-        if not (ddb := Path(ddb).resolve()).exists():
-            raise FileNotFoundError(ddb)
-
-        if not (obsinst := data_dir_ / f"{obsid}.obs").exists():
-            raise FileNotFoundError(obsinst)
-
-        if not (antenna := data_dir_ / f"{obsid}.ant").exists():
-            antenna = None
-
-        if not (cabin := data_dir_ / f"{obsid}.cabin").exists():
-            cabin = None
-
-        if not (misti := data_dir_ / f"{obsid}.misti").exists():
-            misti = None
-
-        if not (skychop := data_dir_ / f"{obsid}.skychopper.dat.xz").exists():
-            skychop = None
-
-        if not (weather := data_dir_ / f"{obsid}.wea").exists():
-            weather = None
-
         return merge.merge(
-            dems,
+            dems_dir / f"dems_{obsid}.zarr.zip",
             # required datasets
-            corresp=corresp,
+            corresp=data_pack_.corresp,
             ddb=ddb,
-            obsinst=obsinst,
+            obsinst=data_pack_.obsinst,
             readout=readout,
             # optional datasets
-            antenna=antenna,
-            cabin=cabin,
-            misti=misti,
-            skychop=skychop,
-            weather=weather,
+            antenna=data_pack_.antenna,
+            cabin=data_pack_.cabin,
+            misti=data_pack_.misti,
+            skychop=data_pack_.skychop,
+            weather=data_pack_.weather,
             # merge options
             measure=measure,
             overwrite=overwrite,
@@ -152,7 +129,7 @@ def demerge(
         )
 
 
-def cli() -> None:
+def demerge_cli() -> None:
     """Command line interface of the demerge function."""
     basicConfig(
         datefmt="%Y-%m-%d %H:%M:%S",

demerge/data/__init__.py

@@ -0,0 +1,73 @@
+__all__ = ["DataPackage", "parse"]
+
+
+# standard library
+from collections.abc import Iterator
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional, Union
+
+
+# type hints
+PathLike = Union[Path, str]
+
+
+@dataclass
+class DataPackage:
+    """Parsed data package structure."""
+
+    antenna: Optional[Path]
+    """Path of the antenna log (optional)."""
+
+    cabin: Optional[Path]
+    """Path of the cabin log (optional)."""
+
+    corresp: Path
+    """Path of the KID correspondence (required)."""
+
+    misti: Optional[Path]
+    """Path of the MiSTI log (optional)."""
+
+    obsinst: Path
+    """Path of the observation instruction (required)."""
+
+    readout: Path
+    """Path of the KID readout FITS (required)."""
+
+    skychop: Optional[Path]
+    """Path of the sky chopper log (optional)."""
+
+    weather: Optional[Path]
+    """Path of the weather log (optional)."""
+
+
+def first(glob_results: Iterator[Path], /) -> Optional[Path]:
+    """Return the first Path.glob result if exists."""
+    for path in glob_results:
+        return path
+
+
+def parse(data_pack: PathLike, /) -> DataPackage:
+    """Parse a data package (data directory)."""
+    if not (data_pack := Path(data_pack)).exists():
+        raise FileNotFoundError(data_pack)
+
+    if (corresp := first(data_pack.glob("*.json"))) is None:
+        raise FileNotFoundError("KID correspondence (*.json).")
+
+    if (obsinst := first(data_pack.glob("*.obs"))) is None:
+        raise FileNotFoundError(f"Observation instruction (*.obs).")
+
+    if (readout := first(data_pack.glob("*.fits*"))) is None:
+        raise FileNotFoundError(f"KID readout FITS (*.fits).")
+
+    return DataPackage(
+        antenna=first(data_pack.glob("*.ant")),
+        cabin=first(data_pack.glob("*.cabin")),
+        corresp=corresp,
+        misti=first(data_pack.glob("*.misti")),
+        obsinst=obsinst,
+        readout=readout,
+        skychop=first(data_pack.glob("*.skychopper*")),
+        weather=first(data_pack.glob("*.wea")),
+    )

demerge/merge/__init__.py

@@ -134,7 +134,7 @@ def merge(
     return dems.resolve()
 
 
-def cli() -> None:
+def merge_cli() -> None:
     """Command line interface of the merge function."""
     basicConfig(
         datefmt="%Y-%m-%d %H:%M:%S",

demerge/reduce/__init__.py

@@ -25,6 +25,11 @@
 SCRIPTS = Path(__file__).parent / "utils" / "scripts" / "aste"
 
 
+def set_dir(dir: PathLike, /) -> Path:
+    """Resolve a directory."""
+    return Path(dir).expanduser().resolve()
+
+
 @contextmanager
 def set_logger(debug: bool, /) -> Iterator[None]:
     """Temporarily set the level of the module logger."""
@@ -41,45 +46,45 @@ def set_logger(debug: bool, /) -> Iterator[None]:
 
 def reduce(
     *,
-    data_dir: PathLike,
-    reduced_dir: PathLike,
+    data_pack: PathLike,
+    reduced_pack: PathLike,
     overwrite: bool = False,
     debug: bool = False,
 ) -> Path:
-    """Reduce raw data of KID measurements into a single "reduced" FITS.
+    """Reduce KID measurements into a single "reduced" FITS.
 
     Args:
-        data_dir: Path of raw data directory (e.g. ``cosmos_YYYYmmddHHMMSS``).
-        reduced_dir: Path of reduced data directory (e.g. ``reduced_YYYYmmddHHMMSS``).
-        overwrite: If True, ``reduced_dir`` will be overwritten even if it exists.
+        data_pack: Path of data package (e.g. ``cosmos_YYYYmmddHHMMSS``).
+        reduced_pack: Path of reduced package (e.g. ``reduced_YYYYmmddHHMMSS``).
+        overwrite: If True, ``reduced_pack`` will be overwritten even if it exists.
         debug: If True, detailed logs for debugging will be printed.
 
     Returns:
-        Path of the created reduced FITS (in the output directory).
+        Path of the created reduced FITS (in the reduced package).
 
     Raises:
-        FileNotFoundError: Raised if ``data_dir`` does not exist.
-        FileExistsError: Raised if ``reduced_dir`` exists.
+        FileNotFoundError: Raised if ``data_pack`` does not exist.
+        FileExistsError: Raised if ``reduced_pack`` exists and overwrite is False.
 
     """
     with set_logger(debug):
         for key, val in locals().items():
             LOGGER.debug(f"{key}: {val!r}")
 
     # Resolve paths (must be done before changing working directory)
-    if not (data_dir := Path(data_dir).resolve()).exists():
-        raise FileNotFoundError(data_dir)
+    if not (data_pack := set_dir(data_pack)).exists():
+        raise FileNotFoundError(data_pack)
 
-    if (reduced_dir := Path(reduced_dir).resolve()).exists() and not overwrite:
-        raise FileExistsError(reduced_dir)
+    if (reduced_pack := set_dir(reduced_pack)).exists() and not overwrite:
+        raise FileExistsError(reduced_pack)
 
     if overwrite:
-        rmtree(reduced_dir, ignore_errors=True)
+        rmtree(reduced_pack, ignore_errors=True)
 
     # Run scripts in a temporary directory (to isolate intermediate files)
     with TemporaryDirectory() as work_dir:
         run(
-            ["python", SCRIPTS / "Configure.py", data_dir, reduced_dir],
+            ["python", SCRIPTS / "Configure.py", data_pack, reduced_pack],
             check=True,
             cwd=work_dir,
             # False if logging is implemented
@@ -100,10 +105,10 @@ def reduce(
             capture_output=True,
         )
 
-    return list(reduced_dir.glob("reduced_*.fits"))[0]
+    return list(reduced_pack.glob("*.fits"))[0]
 
 
-def cli() -> None:
+def reduce_cli() -> None:
     """Command line interface of the reduce function."""
     basicConfig(
         datefmt="%Y-%m-%d %H:%M:%S",

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "demerge"
-version = "2024.7.1"
+version = "2024.7.2"
 description = "DESHIMA merge code for observed datasets"
 authors = [
     "Tatsuya Takekoshi <[email protected]>",
@@ -46,9 +46,9 @@ pytest = "^8.2"
 sphinx = "^7.3"
 
 [tool.poetry.scripts]
-demerge = "demerge:cli"
-merge = "demerge.merge:cli"
-reduce = "demerge.reduce:cli"
+demerge = "demerge:demerge_cli"
+merge = "demerge.merge:merge_cli"
+reduce = "demerge.reduce:reduce_cli"
 
 [tool.black]
 exclude = "demerge/reduce/utils"