Source code for majis.itl.json.reader

"""Majis ITL in JSON format sub-module."""

import json
from pathlib import Path

from planetary_coverage.events import EventWindow

from ...schema import MAJIS_SCHEMAS
from .mapping import JSON_MAPPING




def read_itl_json(
    fname: str | Path,
    fmt: str = 'ITL',
    only: str | list | bool | None = 'MAJIS',
    validation: bool = True,
) -> list[EventWindow]:
    """Read ITL file in JSON format.

    ITL are validated against Juice SOC and Majis schema.

    By default only the MAJIS events will be reported.
    This can be changed by providing an other instrument filter key,
    a list of keys or an explicit 'ALL', False or None value.

    """
    content = json.loads(Path(fname).read_text())

    for schema in MAJIS_SCHEMAS[fmt]:
        schema.validate(content, raises_error=validation)

    # If `filename` is present in content header check that it matches the file name
    if 'filename' in content.get('header', {}):
        filename = Path(content['header']['filename'])

        if filename.name != Path(fname).name:
            raise FileNotFoundError(filename)

    else:
        filename = Path(fname).name

    # Set default values
    defaults = {
        fmt: Path(filename),
    }

    return [
        EventWindow(
            obs['name'],
            parse_json(obs, defaults=defaults),
        )
        for obs in content['timeline']
        # Filter observations by instrument name(s)
        if not only
        or obs['instrument'] in only
        or obs['instrument'] == only
        or only == 'ALL'
    ]





def parse_json(obs: dict, defaults: dict | None = None) -> dict:
    """Map ITL JSON timeline keys to EPS event dictionary."""
    # Extract parameters if present
    params = obs.pop('parameters') if 'parameters' in obs else {}

    # Convert JSON keys to EPS keys
    content = {
        JSON_MAPPING.get(key, key.upper()): value for key, value in (obs | params).items()
    }

    # Add extra keys
    if 'OBSERVATION_TYPE' in content:
        content['PRIME'] = content['OBSERVATION_TYPE'] == 'PRIME'

    # Add default keys if not already present
    if defaults:
        for key, value in defaults.items():
            if key.upper() not in content:
                content[key.upper()] = value

    # Put comments last if present
    content['COMMENTS'] = content.pop('COMMENTS') if 'COMMENTS' in content else ''

    return content