GRM Parser

pynasonde.digisonde.parsers.grm

GRM archive splitter for Digisonde batch files from GIRO DIDBase.

GRM files are raw concatenations of 4096-byte ionogram blocks downloaded from GIRO DIDBase. Each ionogram type is identified by block[0] (the record-type byte):

RSF → 7   SBF → 3   MMM → 9

This module exposes :class:GrmSplitter, which can:

• auto-detect the ionogram format from the file itself
• split the archive to individual files on disk (parallel-safe)
• load each ionogram as a lazily-constructed parser object
• load_dataframes – extract all ionograms and return a flat DataFrame (parallel-safe)

Parallelism strategy

All three methods use a two-phase design:

1. _scan_offsets() — sequential linear scan that records only (datetime, byte_offset, byte_length) per ionogram. No ionogram data is copied.

2. Worker functions — parallel, each opens the file independently and seek()-s to its own slice, so no large bytes objects are pickled across process/thread boundaries.

3. split() → I/O-bound → :class:ThreadPoolExecutor

4. load_dataframes() → CPU-bound → :class:ProcessPoolExecutor
5. load() → creates lazy wrappers, stays sequential

Usage::

spl = GrmSplitter("AU930_20171470000.GRM")
print(spl.fmt)                               # "MMM"

paths = spl.split("/tmp/out/")               # sequential
paths = spl.split("/tmp/out/", n_workers=4)  # parallel (threads)

df = spl.load_dataframes()                   # sequential
df = spl.load_dataframes(n_workers=4)        # parallel (processes)

objs = spl.load()                            # list[_LazyExtractor]
objs[0].extract(); objs[0].to_pandas()

GrmSplitter

Split or load a GRM batch archive from GIRO DIDBase.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Path to the .GRM file.	required
fmt	Optional[str]	Force a specific format ("RSF", "SBF", "MMM"). If None (default) the format is auto-detected.	None
station	Optional[str]	Five-character station code used in output filenames. Defaults to the first five characters of the filename.	None

Source code in pynasonde/digisonde/parsers/grm.py

class GrmSplitter:
    """Split or load a GRM batch archive from GIRO DIDBase.

    Args:
        path:      Path to the ``.GRM`` file.
        fmt:       Force a specific format (``"RSF"``, ``"SBF"``, ``"MMM"``).
                   If ``None`` (default) the format is auto-detected.
        station:   Five-character station code used in output filenames.
                   Defaults to the first five characters of the filename.
    """

    def __init__(
        self,
        path: Union[str, Path],
        fmt: Optional[str] = None,
        station: Optional[str] = None,
    ):
        """Create a GRM archive splitter.

        Args:
            path: Path to the ``.GRM`` file.
            fmt: Optional forced format, one of ``"RSF"``, ``"SBF"``, or
                ``"MMM"``. If omitted, the format is auto-detected.
            station: Station code used when writing split files.
        """
        self.path = Path(path)
        self._fmt: Optional[str] = fmt.upper() if fmt else None
        self.station: str = station or self.path.stem[:5]
        self._offsets: Optional[List[_Slice]] = None  # cached after first scan

    # ── Format detection ──────────────────────────────────────────────────────

    @property
    def fmt(self) -> str:
        """Detected or user-specified ionogram format."""
        if self._fmt is None:
            self._fmt = self.detect_format(self.path)
        return self._fmt

    @staticmethod
    def detect_format(path: Union[str, Path]) -> str:
        """Scan the first matching block and return the format string.

        Raises:
            ValueError: if no known record-type byte is found.
        """
        with open(path, "rb") as f:
            while True:
                block = f.read(BLOCK_SIZE)
                if not block:
                    break
                rt = block[0]
                if rt in _RECORD_TYPE_INV:
                    fmt = _RECORD_TYPE_INV[rt]
                    logger.info(f"Auto-detected format: {fmt} (record_type={rt})")
                    return fmt
        raise ValueError(
            f"Could not detect ionogram format in {path}. "
            f"Expected record_type in {list(_RECORD_TYPE_INV)}"
        )

    # ── Phase 1: sequential offset scan ──────────────────────────────────────

    def _scan_offsets(self) -> List[_Slice]:
        """Scan the GRM file once and return (date, start, length) per ionogram.

        Result is cached — subsequent calls are free.
        """
        if self._offsets is not None:
            return self._offsets

        fmt = self.fmt
        target_rt = _RECORD_TYPE[fmt]
        starts: List[Tuple[dt.datetime, int]] = []  # (date, byte_offset)

        with open(self.path, "rb") as f:
            file_size = f.seek(0, 2)
            f.seek(0)
            offset = 0
            while True:
                block = f.read(BLOCK_SIZE)
                if not block:
                    break
                if block[0] == target_rt:
                    starts.append((_parse_block_date(block, fmt), offset))
                offset += BLOCK_SIZE

        slices: List[_Slice] = []
        for i, (date, start) in enumerate(starts):
            end = starts[i + 1][1] if i + 1 < len(starts) else file_size
            slices.append((date, start, end - start))

        logger.info(
            f"Scanned {self.path.name}: {len(slices)} {fmt} ionograms "
            f"({file_size / 1024 / 1024:.1f} MB)"
        )
        self._offsets = slices
        return slices

    # ── Public methods ────────────────────────────────────────────────────────

    def split(
        self,
        out_dir: Union[str, Path],
        n_workers: int = 1,
    ) -> List[Path]:
        """Write each ionogram to a separate file in *out_dir*.

        Files are named ``<station>_<YYYYDDDHHMMSS>.<fmt>``.

        Args:
            out_dir:    Output directory (created if it does not exist).
            n_workers:  Number of threads for parallel file writes.
                        ``1`` (default) → sequential.
                        ``-1`` → use all CPU cores.

        Returns:
            List of :class:`~pathlib.Path` for every file written,
            in ionogram order.
        """
        out_dir = Path(out_dir)
        out_dir.mkdir(parents=True, exist_ok=True)
        fmt = self.fmt
        slices = self._scan_offsets()
        n_workers = _resolve_workers(n_workers)

        args_list = [
            (str(self.path), self.station, fmt, date, start, length, str(out_dir))
            for date, start, length in slices
        ]

        if n_workers == 1:
            written = [_worker_split(a) for a in args_list]
        else:
            written = [None] * len(args_list)
            with ThreadPoolExecutor(max_workers=n_workers) as pool:
                futures = {pool.submit(_worker_split, a): i for i, a in enumerate(args_list)}
                for fut in as_completed(futures):
                    written[futures[fut]] = fut.result()

        logger.info(f"split → {len(written)} files in {out_dir}")
        return written

    def load(self) -> List["_LazyExtractor"]:
        """Return lazy extractor wrappers — one per ionogram.

        Each object is cheap to construct.  Call ``.extract()`` and
        ``.to_pandas()`` to materialise the data.  Loading is always
        sequential since the wrappers themselves hold no data.

        Returns:
            Ordered list of :class:`_LazyExtractor` objects.
        """
        fmt = self.fmt
        cls = _EXTRACTOR[fmt]
        slices = self._scan_offsets()
        objs = [
            _LazyExtractor(cls, str(self.path), start, length, date)
            for date, start, length in slices
        ]
        logger.info(f"load → {len(objs)} lazy {fmt} extractors")
        return objs

    def load_dataframes(self, n_workers: int = 1) -> pd.DataFrame:
        """Extract all ionograms and return a single concatenated DataFrame.

        Args:
            n_workers:  Number of processes for parallel extraction.
                        ``1`` (default) → sequential.
                        ``-1`` → use all CPU cores.

        Returns:
            Combined :class:`~pandas.DataFrame`.  Empty if nothing extracted.
        """
        fmt = self.fmt
        slices = self._scan_offsets()
        n_workers = _resolve_workers(n_workers)

        # Pass cls as a string key so the tuple is always picklable
        args_list = [
            (str(self.path), fmt, date, start, length)
            for date, start, length in slices
        ]

        if n_workers == 1:
            frames = [_worker_extract(a) for a in args_list]
        else:
            frames = [None] * len(args_list)
            with ProcessPoolExecutor(max_workers=n_workers) as pool:
                futures = {pool.submit(_worker_extract, a): i for i, a in enumerate(args_list)}
                for fut in as_completed(futures):
                    frames[futures[fut]] = fut.result()

        good = [f for f in frames if f is not None and not f.empty]
        if not good:
            logger.warning("No ionograms successfully extracted.")
            return pd.DataFrame()

        result = pd.concat(good, ignore_index=True)
        logger.info(
            f"load_dataframes → {len(good)}/{len(slices)} ionograms, "
            f"{len(result)} total rows"
        )
        return result

fmt: str property

Detected or user-specified ionogram format.

__init__(path, fmt=None, station=None)

Create a GRM archive splitter.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Path to the .GRM file.	required
fmt	Optional[str]	Optional forced format, one of "RSF", "SBF", or "MMM". If omitted, the format is auto-detected.	None
station	Optional[str]	Station code used when writing split files.	None

Source code in pynasonde/digisonde/parsers/grm.py

def __init__(
    self,
    path: Union[str, Path],
    fmt: Optional[str] = None,
    station: Optional[str] = None,
):
    """Create a GRM archive splitter.

    Args:
        path: Path to the ``.GRM`` file.
        fmt: Optional forced format, one of ``"RSF"``, ``"SBF"``, or
            ``"MMM"``. If omitted, the format is auto-detected.
        station: Station code used when writing split files.
    """
    self.path = Path(path)
    self._fmt: Optional[str] = fmt.upper() if fmt else None
    self.station: str = station or self.path.stem[:5]
    self._offsets: Optional[List[_Slice]] = None  # cached after first scan

detect_format(path) staticmethod

Scan the first matching block and return the format string.

Raises:

Type	Description
ValueError	if no known record-type byte is found.

Source code in pynasonde/digisonde/parsers/grm.py

@staticmethod
def detect_format(path: Union[str, Path]) -> str:
    """Scan the first matching block and return the format string.

    Raises:
        ValueError: if no known record-type byte is found.
    """
    with open(path, "rb") as f:
        while True:
            block = f.read(BLOCK_SIZE)
            if not block:
                break
            rt = block[0]
            if rt in _RECORD_TYPE_INV:
                fmt = _RECORD_TYPE_INV[rt]
                logger.info(f"Auto-detected format: {fmt} (record_type={rt})")
                return fmt
    raise ValueError(
        f"Could not detect ionogram format in {path}. "
        f"Expected record_type in {list(_RECORD_TYPE_INV)}"
    )

split(out_dir, n_workers=1)

Write each ionogram to a separate file in out_dir.

Files are named <station>_<YYYYDDDHHMMSS>.<fmt>.

Parameters:

Name	Type	Description	Default
out_dir	Union[str, Path]	Output directory (created if it does not exist).	required
n_workers	int	Number of threads for parallel file writes. 1 (default) → sequential. -1 → use all CPU cores.	1

Returns:

Type	Description
List[Path]	List of :class:~pathlib.Path for every file written,
List[Path]	in ionogram order.

Source code in pynasonde/digisonde/parsers/grm.py

def split(
    self,
    out_dir: Union[str, Path],
    n_workers: int = 1,
) -> List[Path]:
    """Write each ionogram to a separate file in *out_dir*.

    Files are named ``<station>_<YYYYDDDHHMMSS>.<fmt>``.

    Args:
        out_dir:    Output directory (created if it does not exist).
        n_workers:  Number of threads for parallel file writes.
                    ``1`` (default) → sequential.
                    ``-1`` → use all CPU cores.

    Returns:
        List of :class:`~pathlib.Path` for every file written,
        in ionogram order.
    """
    out_dir = Path(out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    fmt = self.fmt
    slices = self._scan_offsets()
    n_workers = _resolve_workers(n_workers)

    args_list = [
        (str(self.path), self.station, fmt, date, start, length, str(out_dir))
        for date, start, length in slices
    ]

    if n_workers == 1:
        written = [_worker_split(a) for a in args_list]
    else:
        written = [None] * len(args_list)
        with ThreadPoolExecutor(max_workers=n_workers) as pool:
            futures = {pool.submit(_worker_split, a): i for i, a in enumerate(args_list)}
            for fut in as_completed(futures):
                written[futures[fut]] = fut.result()

    logger.info(f"split → {len(written)} files in {out_dir}")
    return written

load()

Return lazy extractor wrappers — one per ionogram.

Each object is cheap to construct. Call .extract() and .to_pandas() to materialise the data. Loading is always sequential since the wrappers themselves hold no data.

Returns:

Type	Description
List[_LazyExtractor]	Ordered list of :class:_LazyExtractor objects.

Source code in pynasonde/digisonde/parsers/grm.py

def load(self) -> List["_LazyExtractor"]:
    """Return lazy extractor wrappers — one per ionogram.

    Each object is cheap to construct.  Call ``.extract()`` and
    ``.to_pandas()`` to materialise the data.  Loading is always
    sequential since the wrappers themselves hold no data.

    Returns:
        Ordered list of :class:`_LazyExtractor` objects.
    """
    fmt = self.fmt
    cls = _EXTRACTOR[fmt]
    slices = self._scan_offsets()
    objs = [
        _LazyExtractor(cls, str(self.path), start, length, date)
        for date, start, length in slices
    ]
    logger.info(f"load → {len(objs)} lazy {fmt} extractors")
    return objs

load_dataframes(n_workers=1)

Extract all ionograms and return a single concatenated DataFrame.

Parameters:

Name	Type	Description	Default
n_workers	int	Number of processes for parallel extraction. 1 (default) → sequential. -1 → use all CPU cores.	1

Returns:

Name	Type	Description
Combined	pd.DataFrame	class:~pandas.DataFrame. Empty if nothing extracted.

Source code in pynasonde/digisonde/parsers/grm.py

def load_dataframes(self, n_workers: int = 1) -> pd.DataFrame:
    """Extract all ionograms and return a single concatenated DataFrame.

    Args:
        n_workers:  Number of processes for parallel extraction.
                    ``1`` (default) → sequential.
                    ``-1`` → use all CPU cores.

    Returns:
        Combined :class:`~pandas.DataFrame`.  Empty if nothing extracted.
    """
    fmt = self.fmt
    slices = self._scan_offsets()
    n_workers = _resolve_workers(n_workers)

    # Pass cls as a string key so the tuple is always picklable
    args_list = [
        (str(self.path), fmt, date, start, length)
        for date, start, length in slices
    ]

    if n_workers == 1:
        frames = [_worker_extract(a) for a in args_list]
    else:
        frames = [None] * len(args_list)
        with ProcessPoolExecutor(max_workers=n_workers) as pool:
            futures = {pool.submit(_worker_extract, a): i for i, a in enumerate(args_list)}
            for fut in as_completed(futures):
                frames[futures[fut]] = fut.result()

    good = [f for f in frames if f is not None and not f.empty]
    if not good:
        logger.warning("No ionograms successfully extracted.")
        return pd.DataFrame()

    result = pd.concat(good, ignore_index=True)
    logger.info(
        f"load_dataframes → {len(good)}/{len(slices)} ionograms, "
        f"{len(result)} total rows"
    )
    return result