eos.py

""" NASA EOS mission specific data

References:
    1. MODIS Command, Telemetry, Science and Engineering Description,
       Document Number 151840, May 1997, Appendix C
       https://directreadout.sci.gsfc.nasa.gov/documents/satellite_gen/MODIS_UG.pdf
"""
from typing import Union
from enum import Enum
from dataclasses import dataclass


class PktType(Enum):
    Day = 0
    Night = 1
    Eng1 = 2
    Eng2 = 3


class CalType(Enum):
    SD = 0
    SRCA = 1
    BB = 2
    Space = 3


class CalMode(Enum):
    Radiometric = 0
    Spatial = 1
    Spectral = 2
    NonCal = 3


@dataclass
class CalSource:
    type: CalType
    mode: CalMode
    frame_count: int


@dataclass
class EarthSource:
    frame_count: int


packets_per_scan = {
    "day": 3032,
    "night": 326,
}


def _caltype(p):
    v = p.data[0] >> 5 & 0x3
    if v > 3:
        return None
    return CalType(v)


def _calmode(p):
    v = p.data[0] >> 3 & 0x3
    if v > 3:
        return None
    return CalMode(v)


def _mirrorside(p):
    return (bytes(p.secondary_header)[8] & 0x1) + 1


def _scancount(p):
    return bytes(p.secondary_header)[8] >> 1 & 0x7


def _pkttype(p):
    v = bytes(p.secondary_header)[8] >> 4 & 0x7
    if v > 3:
        return None
    return PktType(v)


def _cal_frame_count(p):
    x = int.from_bytes(b"\x00\x00" + p.data[:2], "big")
    return x >> 4 & 0x3F


def _earth_frame_count(p):
    x = int.from_bytes(b"\x00\x00" + p.data[:2], "big")
    return x >> 4 & 0x7FF


@dataclass
class Info:
    type: str
    packet_type: PktType
    mirror_side: int
    scancount: int
    source: Union[EarthSource, CalSource]


def info(p):
    tp = _pkttype(p)
    source = None
    type = "engr"
    if tp in (PktType.Day, PktType.Night):
        if p.data[0] & 0b10000000 == 0:
            type = "earthdata"
            source = EarthSource(frame_count=_earth_frame_count(p))
        else:
            type = "cal"
            source = CalSource(
                type=_caltype(p),
                mode=_calmode(p),
                frame_count=_earth_frame_count(p),
            )
    return Info(type, tp, _mirrorside(p), _scancount(p), source)


def _modis_sort_key(p):
    """Return a tuple that will maintain order for this packet in a stream of
    MODIS data. Intended to be used as a key func for sorting.

    Packets are sorted in the order defined in _sort_indexes.
    """
    frame_sort = 0
    i = info(p)
    if i.type == "earthdata":
        type_idx = 5
        frame_sort = i.source.frame_count
    elif i.type == "cal":
        type_idx = i.source.type.value
        frame_sort = i.source.frame_count
    else:
        type_idx = 6 if i.type == PktType.Eng1 else 7
    return (p.stamp, 64, type_idx, frame_sort, int(p.is_last()))


def sort_modis(packets):
    return sorted(packets, key=_modis_sort_key)