Skip to content

Commit

Permalink
Docs and docstrings
Browse files Browse the repository at this point in the history
  • Loading branch information
taldcroft committed Feb 20, 2025
1 parent c0290c1 commit f198a33
Show file tree
Hide file tree
Showing 2 changed files with 124 additions and 6 deletions.
54 changes: 54 additions & 0 deletions docs/commands_states/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -889,6 +889,60 @@ transition callback functions. There are a number of examples of this in the ka
and this should serve as your starting point. The Ska team will be happy to assist
you if this is not enough.

Filtered Event History
----------------------

In some cases you may want to use a filtered event history, most commonly to see the
originally planned commands and states following an SCS-107 event. Since SCS-107 stops
the observing loads, commands in those loads like the ObsID update and science
instrument commanding stop appearing in the kadi command history and states.

Filtering the Chandra command events is done by specifying the ``event_filter`` argument
to any of |get_cmds|, |get_states|, |get_continuity|,
:func:`~kadi.commands.observations.get_observations`,
:func:`~kadi.commands.observations.get_starcats`, and
:func:`~kadi.commands.observations.get_starcats_as_table`. This filter allows for
dynamically excluding command events from the flight Command Events sheet or whatever
scenario you have chosen.

The ``event_filter`` argument takes a single callable function or a list of functions.
Each function must take a single argument, an Command Events Table object, and return a
boolean mask indicating whether each row (command event) in the table should be included
when generating the final commands. The functions are applied in order, and the command
event is included if any of the functions return ``True``.

Filtering within last 30 days
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
If the events you want to filter occurred within the last 30 days, you can use any of
the supported functions (listed above) directly, for example::

>>> from kadi.commands import filter_scs107_events
>>> obss_planned = get_observations(start, stop, event_filter=filter_scs107_events)

Filtering events older than 30 days
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
By default, the commands archive dynamically applies the event history from the last 30
days when constructing the command history from backstop files. Beyond 30 days, the
commands are taken from a static HDF5 archive file. To filter events older than 30
days, you need to make kadi pretend that the "current time" is actually a date in the
past.

For example, to get the "as-planned" observations covering two SCS-107 runs in the
beginning of 2025, you can use the following code. This uses the
:func:`~kadi.commands.set_time_now` context manager to temporarily set the current time
to a date in the past, and the :func:`~kadi.commands.filter_scs107_events` function to
filter out the SCS-107 events::

>>> from kadi.commands import filter_scs107_events, set_time_now
>>> with set_time_now("2025:015"):
... obss_planned = get_observations(
... "2024:365", "2025:015", event_filter=filter_scs107_events
... )

For a custom filter you can use :func:`~kadi.commands.filter_cmd_events_by_event`, or
build your own filtering function that takes a command events table as input returns a
boolean numpy array mask.

Commands archive details
------------------------

Expand Down
76 changes: 70 additions & 6 deletions kadi/commands/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -1143,6 +1143,20 @@ def get_cxotime_now() -> str | None:


def filter_cmd_events_date_stop(date_stop):
"""
Returns a function that removes command events with ``Date > date_stop``.
Parameters
----------
date_stop : CxoTimeLike
Date string in CxoTime-compatible format.
Returns
-------
Callable
Function that takes a Table of command events and returns a boolean numpy array.
"""

def func(cmd_events):
stop = CxoTime(date_stop)
# Filter table based on stop date. Need to use CxoTime on each event separately
Expand All @@ -1157,21 +1171,71 @@ def func(cmd_events):
return func


def filter_cmd_events_by_event(event_events: list[str]) -> Callable:
@functools.lru_cache()
def filter_cmd_events_by_event(*args: tuple[str]) -> Callable:
"""
Returns a function that filters command events based on a list of event names.
Example::
>>> import kadi.commands as kc
>>> filter = kc.filter_cmd_events_by_event("SCS-107", "RTS", "Obsid", "Observing not run")
>>> with kc.set_time_now("2025:015"):
... cmds = kc.get_cmds("2025:001", "2025:015", event_filter=filter)
See also: :func:`~kadi.commands.core.filter_scs107_events`.
Parameters
----------
*args : str
Event names (e.g., "SCS-107" or "RTS") to filter out from the command events.
Returns
-------
Callable
Function that takes a Table of command events and returns a boolean numpy array.
"""

def func(cmd_events: Table) -> np.ndarray[bool]:
ok = ~np.isin(cmd_events["Event"], event_events)
ok = ~np.isin(cmd_events["Event"], args)
logger.info(
f"Filtering cmd_events['Event'] that match {event_events} "
f"Filtering cmd_events['Event'] that match {args} "
f"({np.count_nonzero(ok)} vs {len(cmd_events)})"
)
return ok

return func


filter_scs107_events = filter_cmd_events_by_event(SCS107_EVENTS)
filter_scs107_events.__doc__ = "Filter SCS-107 related events from command history."
filter_scs107_events.__name__ = "filter_scs107_events"
def filter_scs107_events(cmd_events: Table) -> np.ndarray[bool]:
"""Filter SCS-107 related events from command history.
This filters out the following command event types:
- "SCS-107"
- "Observing not run"
- "Obsid"
- "RTS"
Example::
>>> import kadi.commands as kc
>>> with kc.set_time_now("2025:015"):
... cmds = kc.get_cmds("2025:001", "2025:015", event_filter=kc.filter_scs107_events)
Parameters
----------
cmd_events : Table
Command events, where each event has an "Event" attribute.
Returns
-------
numpy.ndarray
A boolean array indicating which command events are not SCS-107 related.
"""
filter_func = filter_cmd_events_by_event(*SCS107_EVENTS)
return filter_func(cmd_events)


def filter_cmd_events_state(cmd_events: Table) -> np.ndarray[bool]:
Expand Down

0 comments on commit f198a33

Please sign in to comment.