csh_plots.py

#!/usr/bin/env python

"""
**csh_plots.py**: Use the msid_plotting package to generate recent plots of maude CSH MSID's.

:Author: W. Aaron (william.aaron@cfa.harvad.edu)
:Last Updated: Apr 02, 2026

:NOTE: This script depends on the msid_plotting package but no runtime environment containing this package has been configured.
    Instead, the regular run involves installing a copy of the msid_plotting package in the script directory.
    Test runs of this script can avoid this step by declaring a PYTHONPATH env variable to the msid_plotting git submodule.

# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "msid_plotting>=0.4",
# ]
# ///

# /// testing
# tested-ska-release = "2026.1"
# ///
"""
import os
import json
import argparse
import signal
from typing import Any
from datetime import timedelta
from cxotime import CxoTime
import psutil
import msid_plotting
from msid_plotting import comm_check
from pathlib import Path
#
#--- Define Directory Pathing
#
SOH_DIR = Path(os.getenv("SOH_DIR", "/data/mta4/Script/SOH"))
SOH_WEB_DIR = Path(os.getenv("SOH_WEB_DIR", "/data/mta4/www/CSH"))
SOH_PLOT_DIR = SOH_WEB_DIR / "Plots"

SCRIPT_DIR = Path(__file__).parent
HOUSE_KEEPING = SCRIPT_DIR / "house_keeping"
PLOT_CONFIG_FILE = HOUSE_KEEPING / "plot_configurations.json"

NOW = CxoTime()
BIN_SIZE = 500
RUN = False #: Override the comm check and run regardless
_NCOLS = {
    1:1,
    2:2,
    3:3,
    4:2,
    5:3,
    6:3,
    7:3,
    8:3,
    9:3
}
FETCH_KWARGS = {
    "channel": "FLIGHT", # options (FLIGHT, FLTCOMP, ASVT, TEST)
    #"highrate": True, #High data rate
    #"allpoints": True, #Include all points in the query fetch
    #"include_calcs": True, #include calc-type blobs in spacecraft blob queries
}
def get_options(args=None):
    parser = argparse.ArgumentParser(description="Plot Maude CSH MSIDs")
    parser.add_argument("-m", "--mode", choices = ['flight','test'], required = True, help = "Determine running mode.")
    parser.add_argument("--config", required = False, help="Pass alternative plot configuration JSON file.")
    parser.add_argument("--force-run", action='store_true', help="Force the script to run regardless of being in comm.")
    opt = parser.parse_args(args)
    return opt

def main():
    """
    Check if we are in comm and plot recent maude data.
    """
    #: Pull DSN Comm Information
    comm_info = comm_check.comm_check()
    comm_time_info = comm_check.translate(comm_info['comm'])
    state_info = comm_check.in_state(**comm_time_info)

    #: Format the DSN Comm information as a title annotation to plots
    if state_info['in_track']:
        #: Updating the data in plot right now
        run = True
        comm_annotation = f"Last Updated: {NOW.date}\n"
        comm_annotation += f"Current Comm Ends: {comm_time_info['track_stop'].date}"
    elif comm_time_info['track_stop'] < NOW < comm_time_info['support_stop']:
        #: Just finished comm. Run once more and format the annotation for the next comm
        run = True
        comm_annotation = f"Last Updated: {comm_time_info['track_stop'].date}\n"
        _time_info = comm_check.translate(comm_info['dsn_query'][1])
        comm_annotation += f"Next DSN Comm: {_time_info['track_start'].date}"
    else:
        run = False
        #: Automated run choice is to not run the script,
        #: but generate the annotations in case the global RUN choice is true.
        _time_info = comm_check.translate(comm_info['previous_comm'])
        comm_annotation = f"Last Updated: {_time_info['track_stop'].date}\n"
        comm_annotation += f"Next DSN Comm: {comm_time_info['track_start'].date}"

    if run or RUN:        
        with open(PLOT_CONFIG_FILE) as f:
            plot_configurations = json.load(f)
        
        stop = NOW
        start = NOW - timedelta(days=2)

        #: Add Template Directory
        msid_plotting.msid_plot.JINJA_TEMPLATE_ENV.add_template_directory(
            f"{os.path.dirname(__file__)}/template"
        )

        for category, config in plot_configurations.items():
            msid_ls = config.get('msid_ls')
            title = config.get('title')
            units = config.get('units')
            weight = config.get('weight')

            template_variables = {
                'title': f"Plot: {category}",
                }
            generate_plot(category, msid_ls, start, stop, comm_annotation, title, units, weight, template_variables)

def generate_plot(category, msid_ls, start, stop, comm_annotation = None, title = None, units = None, weight = None, template_variables = {}):
    """
    Generate the plot html subpage for a given MSID set and associated parameters.
    """
    multivar_plot = msid_plotting.msid_plot.MSIDPlot(
        msids = msid_ls,
        start = start,
        stop = stop,
        bin_size = BIN_SIZE
    )

    params : dict[str, Any] = {'size': 5}
    if title is not None:
        params['title'] = title
        if comm_annotation is not None:
            params['title'] += f"\n{comm_annotation}"
    if units is not None:
        params['y_axis_labels'] = [f"{msid}({unit})" for msid, unit in zip(multivar_plot.msids, units)]
    if weight is not None:
        params['weights'] = weight
    
    multivar_plot.parameterize(params)
    multivar_plot.fetch_data(query_maude_kwargs=FETCH_KWARGS)

    html = multivar_plot.generate_plot_html(
        template_name='maude_csh_plot.jinja',
        template_variables = template_variables,
        ncols = _NCOLS[len(msid_ls)]
        )
    file = SOH_PLOT_DIR / f"{category}.html"

    with open(file,'w') as f:
        f.write(html)


if __name__ == "__main__":
    opt = get_options()
    if opt.mode == 'test':
        SOH_PLOT_DIR = Path(os.getcwd(), "test", "_outTest" ,"Plots")
        os.makedirs(SOH_PLOT_DIR, exist_ok = True)
        if opt.config is not None:
            PLOT_CONFIG_FILE = opt.config
        RUN = opt.force_run
        main()

    if opt.mode == 'flight':
            #: Create a lock file and exit strategy in case of race conditions.
        name = os.path.basename(__file__).split(".")[0]
        user = os.getenv("USER", "mta")
        lock = Path("/tmp", user, f"{name}.lock")

        #: If lock file exists, read the pid and kill the process, then remove the lock file
        if os.path.isfile(lock):
            with open(lock) as f:
                pid = int(f.read().strip())
            if psutil.pid_exists(pid):
                os.kill(pid, signal.SIGTERM)
            os.remove(lock)
        
        #: Lock file with current pid
        pid = os.getpid()
        os.makedirs(os.path.dirname(lock), exist_ok = True)
        with open(lock, 'w') as f:
            f.write(str(pid))

        main()

        #: Remove lock file once process is completed
        os.remove(lock)