diff --git a/.github/workflows/changelog.yaml b/.github/workflows/changelog.yaml new file mode 100644 index 0000000..23172c3 --- /dev/null +++ b/.github/workflows/changelog.yaml @@ -0,0 +1,11 @@ +name: ChangelogUpdated + +on: + pull_request: + types: [assigned, opened, synchronize, reopened, labeled, unlabeled] + branches: + - develop + +jobs: + call-workflow: + uses: lsst-ts/tssw_workflows/.github/workflows/news_creation.yaml@main diff --git a/.ts_pre_commit_config.yaml b/.ts_pre_commit_config.yaml index ffdff69..429e368 100644 --- a/.ts_pre_commit_config.yaml +++ b/.ts_pre_commit_config.yaml @@ -1,7 +1,11 @@ -check-yaml: true +black: false check-xml: true -black: true -flake8: true -isort: true -mypy: false +check-yaml: true +clang-format: false +flake8: false +format-xmllint: false +isort: false +mypy: true +nbstripout: false ruff: true +towncrier: true diff --git a/doc/conf.py b/doc/conf.py index ab6cf24..45b8b5b 100644 --- a/doc/conf.py +++ b/doc/conf.py @@ -3,14 +3,12 @@ This configuration only affects single-package Sphinx documentation builds. """ -import lsst.ts.scriptqueue # noqa -from documenteer.conf.pipelinespkg import * # type: ignore # noqa +from documenteer.conf.guide import * # type: ignore # noqa project = "ts_scriptqueue" html_theme_options["logotext"] = project # type: ignore # noqa html_title = project html_short_title = project -doxylink = {} # Avoid warning: Could not find tag file _doxygen/doxygen.tag intersphinx_mapping["ts_salobj"] = ("https://ts-salobj.lsst.io", None) # type: ignore # noqa intersphinx_mapping["ts_utils"] = ("https://ts-utils.lsst.io", None) # type: ignore # noqa diff --git a/doc/documenteer.toml b/doc/documenteer.toml new file mode 100644 index 0000000..1d2757e --- /dev/null +++ b/doc/documenteer.toml @@ -0,0 +1,15 @@ + +[project] +title = "ScriptQueue CSC." +copyright = "2015-2024 Association of Universities for Research in Astronomy, Inc. (AURA)" +github_url = "https://github.com/lsst-ts/ts_scriptqueue" + +[project.python] +package = "ts_scriptqueue" +documentation_url_key = "documentation" +github_url_key = "repository" + +[sphinx.intersphinx.projects] +ts_xml = "https://ts-xml.lsst.io" +ts_salobj = "https://ts-salobj.lsst.io" +python = "https://docs.python.org/3.13/" diff --git a/doc/news/OSW-350.doc.rst b/doc/news/OSW-350.doc.rst new file mode 100644 index 0000000..58fb720 --- /dev/null +++ b/doc/news/OSW-350.doc.rst @@ -0,0 +1 @@ +Update documentation build. diff --git a/doc/news/OSW-350.feature.rst b/doc/news/OSW-350.feature.rst new file mode 100644 index 0000000..0a5549f --- /dev/null +++ b/doc/news/OSW-350.feature.rst @@ -0,0 +1 @@ +Added estimated start time to the next visit event in script_queue.py. diff --git a/doc/news/OSW-350.misc.rst b/doc/news/OSW-350.misc.rst new file mode 100644 index 0000000..9dd501c --- /dev/null +++ b/doc/news/OSW-350.misc.rst @@ -0,0 +1 @@ +Added support for type checking with mypy. diff --git a/doc/news/README.rst b/doc/news/README.rst new file mode 100644 index 0000000..20f1ffd --- /dev/null +++ b/doc/news/README.rst @@ -0,0 +1,42 @@ +Recording Changes +================= + +This directory contains "news fragments" which are small, structured text files that contain information about changes or updates that will be included in the release notes. +These fragments are used to automatically generate changelogs or release notes. +They can be written restructured text format or plain text. + +Each file should be named like ``..`` with a file extension defining the markup format (``rst|md``). +The ```` should be one of: + +* ``feature``: A new feature +* ``bugfix``: A bug fix. +* ``perf``: A performance enhancement. +* ``doc``: A documentation improvement. +* ``removal``: A deprecation or removal of API. +* ``misc``: Other minor changes and/or additions + +An example file name would therefore look like ``DM-40534.doc.rst``. + +Each developer now has to create the news fragments for the changes they have made on their own branches, +instead of adding them to the release notes directly. +The news fragments are then automatically integrated into the release notes by the ``towncrier`` tool. + +You can test how the content will be integrated into the release notes by running ``towncrier build --draft --version=v``. +Note that you have to run it from the root repository directory (i.e. the ``ts_scriptqueue``). + +In order to update the release notes file for real, the person responsible for the releasing the notes should run: + +.. code-block:: bash + + $ towncrier build --version=v + + +.. note:: + + When running towncrier to build the changelog, you may be prompted to confirm the deletion of fragments. + If you would like to retain the fragments in the doc/news directory do not confirm the deletion. + +Note also that ``towncrier`` can be installed from PyPI or conda-forge. + + + diff --git a/doc/requirements.txt b/doc/requirements.txt new file mode 100644 index 0000000..3acd4c0 --- /dev/null +++ b/doc/requirements.txt @@ -0,0 +1,3 @@ +documenteer[pipelines,guide] +sphinxext-rediraffe + diff --git a/doc/version_history.rst b/doc/version_history.rst index 76aa154..599d72a 100644 --- a/doc/version_history.rst +++ b/doc/version_history.rst @@ -6,6 +6,8 @@ Version History ############### +.. towncrier release notes start + v2.14.2 ------- diff --git a/python/lsst/ts/scriptqueue/block_info.py b/python/lsst/ts/scriptqueue/block_info.py index 7730026..795c573 100644 --- a/python/lsst/ts/scriptqueue/block_info.py +++ b/python/lsst/ts/scriptqueue/block_info.py @@ -21,12 +21,15 @@ __all__ = ["BlockInfo"] +import logging import os import re from collections import deque from lsst.ts.utils import ImageNameServiceClient +from .type_hints import ScriptInfoProtocol + BLOCK_REGEX = re.compile(r"(?PBLOCK-T)?(?PBLOCK-)?(?P[0-9]*)") @@ -47,20 +50,20 @@ class BlockInfo: How many scripts are part of this block. """ - def __init__(self, log, block_id, block_size): + def __init__(self, log: logging.Logger, block_id: str, block_size: int) -> None: self.log = log.getChild("BlockInfo") self.block_id = block_id self.block_size = block_size block_match = BLOCK_REGEX.match(block_id) - if block_match.span()[1] == 0: + if block_match is None or block_match.span()[1] == 0: raise ValueError(f"{block_id} has the wrong format, should be BLOCK-N or BLOCK-TN.") self._block_ticket_id = abs(int(block_match.groupdict()["id"])) self._block_type = "BlockT" if block_match.groupdict()["block_test_case"] is not None else "Block" - self._block_uid = None - self.scripts_info = deque(maxlen=block_size) + self._block_uid: str | None = None + self.scripts_info: deque[ScriptInfoProtocol] = deque(maxlen=int(block_size)) self.image_server_url = os.environ.get("IMAGE_SERVER_URL") if self.image_server_url is None: @@ -69,7 +72,7 @@ def __init__(self, log, block_id, block_size): "Block indexing functionality will not work." ) - def get_block_uid(self): + def get_block_uid(self) -> str: """Retrieve block uid. Returns @@ -77,12 +80,12 @@ def get_block_uid(self): block_uid : `str` Block unique id. """ - if not self.has_uid(): + if self._block_uid is None: raise RuntimeError("Block uid has not been set yet, call set_block_uid first.") return self._block_uid - def has_uid(self): + def has_uid(self) -> bool: """Check if block uid was set. Returns @@ -92,7 +95,7 @@ def has_uid(self): """ return self._block_uid is not None - async def set_block_uid(self): + async def set_block_uid(self) -> None: """Retrieve and set the block unique id from the name server.""" if self._block_uid is not None: @@ -105,12 +108,12 @@ async def set_block_uid(self): _, data = await image_server_client.get_next_obs_id(num_images=1) self._block_uid = data[0] - def add(self, script_info): + def add(self, script_info: ScriptInfoProtocol) -> None: """Add Script to the block. Parameters ---------- - script_info : `ScriptInfo` + script_info : `ScriptInfoProtocol` ScriptInfo for the script to add to the block. """ if self._block_uid is None: @@ -127,7 +130,7 @@ def add(self, script_info): self.scripts_info.append(script_info) script_info.set_block_index(index) - def done(self): + def done(self) -> bool: """Check if block is done. A block is considered done is all the scripts that diff --git a/python/lsst/ts/scriptqueue/block_model.py b/python/lsst/ts/scriptqueue/block_model.py index 71a7395..9ba7929 100644 --- a/python/lsst/ts/scriptqueue/block_model.py +++ b/python/lsst/ts/scriptqueue/block_model.py @@ -23,6 +23,8 @@ "BlockModel", ] +from .block_info import BlockInfo + class BlockModel: """Manages block information. @@ -34,11 +36,11 @@ class BlockModel: fails the entire block must fail. """ - def __init__(self): - self.blocks = dict() - self.current_blocks = dict() + def __init__(self) -> None: + self.blocks: dict[str, dict[str, BlockInfo]] = dict() + self.current_blocks: dict[str, str] = dict() - def add_block(self, block_info): + def add_block(self, block_info: BlockInfo) -> None: """Add block info to the list of blocks. When a new block is added it will become the "current" @@ -60,7 +62,7 @@ def add_block(self, block_info): self.current_blocks[block_info.block_id] = block_info.get_block_uid() - def get_current_block(self, block_id): + def get_current_block(self, block_id: str) -> BlockInfo: """Return the BlockInfo for the current block. Parameters @@ -81,7 +83,7 @@ def get_current_block(self, block_id): return block_info - def remove_done_blocks(self): + def remove_done_blocks(self) -> None: """Remove blocks that have already finished.""" for block in self.blocks: diff --git a/python/lsst/ts/scriptqueue/queue_model.py b/python/lsst/ts/scriptqueue/queue_model.py index 6e4d443..d215b81 100644 --- a/python/lsst/ts/scriptqueue/queue_model.py +++ b/python/lsst/ts/scriptqueue/queue_model.py @@ -25,21 +25,25 @@ import collections import copy import inspect +import logging import os import pathlib import signal import astropy.time import psutil + from lsst.ts import salobj from lsst.ts.utils import index_generator from lsst.ts.xml.enums.Script import ScriptState from lsst.ts.xml.enums.ScriptQueue import Location +from lsst.ts.xml.type_hints import BaseMsgType from . import utils from .block_info import BlockInfo from .block_model import BlockModel from .script_info import ScriptInfo +from .type_hints import AsyncNoArgsCallback, AsyncScriptInfoBoolCallback, AsyncScriptInfoCallback, Indexed # Standard timeout (seconds). Long enough to perform any reasonable operation, # including starting a CSC or loading a script (seconds) @@ -60,7 +64,7 @@ class Scripts: Relative paths to external SAL scripts """ - def __init__(self, standard, external): + def __init__(self, standard: list[str], external: list[str]) -> None: self.standard = standard self.external = external @@ -75,19 +79,23 @@ class ScriptKey: components that are currently running. """ - def __init__(self, index): + def __init__(self, index: int) -> None: self.index = int(index) - def __hash__(self): + def __hash__(self) -> int: return self.index - def __eq__(self, other): + def __eq__(self, other: object) -> bool: + if not isinstance(other, Indexed): + return NotImplemented return self.index == other.index - def __ne__(self, other): + def __ne__(self, other: object) -> bool: + if not isinstance(other, Indexed): + return NotImplemented return not (self == other) - def __repr__(self): + def __repr__(self) -> str: return f"ScriptKey(index={self.index})" @@ -135,17 +143,17 @@ class QueueModel: def __init__( self, - domain, - log, - standardpath, - externalpath, - next_visit_callback=None, - next_visit_canceled_callback=None, - queue_callback=None, - script_callback=None, - min_sal_index=MIN_SAL_INDEX, - max_sal_index=salobj.MAX_SAL_INDEX, - verbose=False, + domain: salobj.Domain, + log: logging.Logger, + standardpath: os.PathLike, + externalpath: os.PathLike, + next_visit_callback: AsyncScriptInfoCallback | None = None, + next_visit_canceled_callback: AsyncScriptInfoCallback | None = None, + queue_callback: AsyncNoArgsCallback | None = None, + script_callback: AsyncScriptInfoBoolCallback | None = None, + min_sal_index: int = MIN_SAL_INDEX, + max_sal_index: int = salobj.MAX_SAL_INDEX, + verbose: bool = False, ): if not os.path.isdir(standardpath): raise ValueError(f"No such dir standardpath={standardpath}") @@ -173,14 +181,14 @@ def __init__( self.max_sal_index = max_sal_index self.verbose = verbose # queue of ScriptInfo instances - self.queue = collections.deque() - self.history = collections.deque(maxlen=MAX_HISTORY) + self.queue: collections.deque[ScriptInfo] = collections.deque() + self.history: collections.deque[ScriptInfo] = collections.deque(maxlen=MAX_HISTORY) self.block_model = BlockModel() - self.current_script = None + self.current_script: ScriptInfo | None = None self._running = True self._enabled = False self._index_generator = index_generator(imin=min_sal_index, imax=max_sal_index) - self._scripts_being_stopped = set() + self._scripts_being_stopped: set[int] = set() # use index=0 so we get messages for all scripts self.remote = salobj.Remote(domain=domain, name="Script", index=0, evt_max_history=0) self.remote.evt_metadata.callback = self._script_metadata_callback @@ -191,12 +199,12 @@ def __init__( async def add( self, - script_info, - location, - location_sal_index, - start_block=False, - block_size=0, - ): + script_info: ScriptInfo, + location: Location, + location_sal_index: int, + start_block: bool = False, + block_size: int = 0, + ) -> None: """Add a script to the queue. Launch the script in a new subprocess and wait for the subprocess @@ -249,25 +257,25 @@ async def add( await asyncio.wait_for(coro, timeout=STD_TIMEOUT) @property - def current_index(self): + def current_index(self) -> int: """SAL index of the current script, or 0 if none.""" return 0 if self.current_script is None else self.current_script.index @property - def history_indices(self): + def history_indices(self) -> list[int]: """SAL indices of scripts on the history queue.""" return [script_info.index for script_info in self.history] @property - def queue_indices(self): + def queue_indices(self) -> list[int]: """SAL indices of scripts on the queue.""" return [script_info.index for script_info in self.queue] - async def close(self): + async def close(self) -> None: """Shut down the queue, terminate all scripts and free resources.""" await self.terminate_all() - def find_available_scripts(self): + def find_available_scripts(self) -> Scripts: """Find available scripts. Returns @@ -280,7 +288,7 @@ def find_available_scripts(self): external=utils.find_public_scripts(self.externalpath), ) - def get_queue_index(self, sal_index): + def get_queue_index(self, sal_index: int) -> int: """Get queue index of a script on the queue. Parameters @@ -294,9 +302,9 @@ def get_queue_index(self, sal_index): If the script cannot be found on the queue. """ key = ScriptKey(sal_index) - return self.queue.index(key) + return self.queue.index(key) # type: ignore[arg-type] - def get_script_info(self, sal_index, search_history): + def get_script_info(self, sal_index: int, search_history: bool) -> ScriptInfo: """Get information about a script. Search current script, the queue and history. @@ -305,6 +313,14 @@ def get_script_info(self, sal_index, search_history): ---------- sal_index : `int` SAL index of script. + search_history : `bool` + Search past history? If False will only look for the current + and queued scripts. + + Returns + ------- + `ScriptInfo` + Script info for the requested sal index. Raises ------ @@ -315,15 +331,17 @@ def get_script_info(self, sal_index, search_history): return self.current_script key = ScriptKey(sal_index) try: - return self.queue[self.queue.index(key)] + queue_index = self.queue.index(key) # type: ignore[arg-type] + return self.queue[queue_index] except ValueError: if search_history: pass else: raise - return self.history[self.history.index(key)] + history_index = self.history.index(key) # type: ignore[arg-type] + return self.history[history_index] - def make_full_path(self, is_standard, path): + def make_full_path(self, is_standard: bool, path: str | os.PathLike) -> os.PathLike: """Make a full path from path and is_standard and check that it points to a runnable script. @@ -363,7 +381,7 @@ def make_full_path(self, is_standard, path): raise ValueError(f"Script {fullpath} is not executable.") return fullpath - async def move(self, sal_index, location, location_sal_index): + async def move(self, sal_index: int, location: Location, location_sal_index: int) -> None: """Move a script within the queue. Parameters @@ -405,11 +423,11 @@ async def move(self, sal_index, location, location_sal_index): raise @property - def next_sal_index(self): + def next_sal_index(self) -> int: """Get the next available SAL Script index.""" return next(self._index_generator) - def pop_script_info(self, sal_index): + def pop_script_info(self, sal_index: int) -> ScriptInfo: """Remove and return information about a script on the queue. Parameters @@ -417,6 +435,11 @@ def pop_script_info(self, sal_index): sal_index : `int` SAL index of script. + Returns + ------- + `ScriptInfo` + Script infor of the removed script. + Raises ------ ValueError @@ -427,7 +450,9 @@ def pop_script_info(self, sal_index): del self.queue[queue_index] return script_info - async def requeue(self, sal_index, seq_num, location, location_sal_index): + async def requeue( + self, sal_index: int, seq_num: int, location: Location, location_sal_index: int + ) -> ScriptInfo: """Requeue a script. Add a script that is a copy of an existing script, @@ -439,8 +464,6 @@ async def requeue(self, sal_index, seq_num, location, location_sal_index): Parameters ---------- - domain : `lsst.ts.salobj.Domain` - DDS domain. sal_index : `int` SAL index of script to requeue. seq_num : `int` @@ -488,7 +511,7 @@ async def requeue(self, sal_index, seq_num, location, location_sal_index): ) return script_info - async def stop_scripts(self, sal_indices, terminate): + async def stop_scripts(self, sal_indices: list[int], terminate: bool) -> None: """Stop one or more queued scripts and/or the current script. Silently ignores scripts that cannot be found or are already stopped. @@ -525,7 +548,7 @@ async def stop_scripts(self, sal_indices, terminate): finally: self._scripts_being_stopped = set() - async def stop_one_script(self, script_info): + async def stop_one_script(self, script_info: ScriptInfo) -> None: """Stop a queued or running script, giving it time to clean up. First send the script the ``stop`` command, giving that ``timeout`` @@ -548,7 +571,8 @@ async def stop_one_script(self, script_info): try: await script_info.remote.cmd_stop.set_start(salIndex=script_info.index, timeout=STD_TIMEOUT) # give the process time to terminate - await asyncio.wait_for(script_info.process.wait(), timeout=STD_TIMEOUT) + if script_info.process is not None: + await asyncio.wait_for(script_info.process.wait(), timeout=STD_TIMEOUT) # let the script be removed or moved await asyncio.sleep(0) return @@ -557,7 +581,7 @@ async def stop_one_script(self, script_info): pass await self.terminate_one_script(script_info) - async def terminate_one_script(self, script_info): + async def terminate_one_script(self, script_info: ScriptInfo) -> None: """Terminate a queued or running script. If successful (as it will be, unless the script catches SIGTERM), @@ -594,14 +618,14 @@ async def terminate_one_script(self, script_info): await asyncio.sleep(0) @property - def enabled(self): + def enabled(self) -> bool: """Get enabled state. True if ScriptQueue is in the enabled state, False otherwise. """ return self._enabled - async def set_enable(self, enabled): + async def set_enable(self, enabled: bool) -> None: """Set enabled state.""" was_enabled = self._enabled self._enabled = bool(enabled) @@ -609,14 +633,14 @@ async def set_enable(self, enabled): await self._update_queue() @property - def running(self): + def running(self) -> bool: """Get or set running state. If set False the queue pauses. """ return self._running - async def set_running(self, run): + async def set_running(self, run: bool) -> None: """Set running state.""" was_running = self._running self._running = bool(run) @@ -624,7 +648,7 @@ async def set_running(self, run): await self._update_queue(pause_on_failure=False) @staticmethod - def next_group_id(): + def next_group_id() -> str: """Get the next group ID. The group ID is the current TAI date and time as a string in ISO @@ -634,7 +658,7 @@ def next_group_id(): """ return astropy.time.Time.now().tai.isot - async def terminate_all(self): + async def terminate_all(self) -> list[ScriptInfo]: """Terminate all scripts and return info for the ones terminated. Returns @@ -669,7 +693,9 @@ async def terminate_all(self): return info_list - async def _insert_script(self, script_info, location, location_sal_index): + async def _insert_script( + self, script_info: ScriptInfo, location: Location, location_sal_index: int + ) -> None: """Insert a script info into the queue. Parameters @@ -707,7 +733,7 @@ async def _insert_script(self, script_info, location, location_sal_index): script_info.callback = self._script_info_callback await self._update_queue() - async def _remove_script(self, sal_index): + async def _remove_script(self, sal_index: int) -> None: """Remove a script from the queue.""" key = ScriptKey(sal_index) self.log.debug(f"Removing script {key} from the queue.") @@ -741,7 +767,7 @@ async def _remove_script(self, sal_index): self.log.debug("Updating queue.") await self._update_queue() - async def _log_message_callback(self, data): + async def _log_message_callback(self, data: BaseMsgType) -> None: """Print Script logMessage data to stdout. To use: if self.verbose is true then set this as a callback @@ -757,7 +783,7 @@ async def _log_message_callback(self, data): f"level={data.level}; traceback={data.traceback!r}" ) - async def clear_group_id(self, script_info, command_script): + async def clear_group_id(self, script_info: ScriptInfo, command_script: bool) -> None: """Clear the group ID of the specified script, if appropriate. Clear the group ID of the specified script if the group ID @@ -781,7 +807,7 @@ async def clear_group_id(self, script_info, command_script): self.log.exception("next_visit_canceled_callback failed; continuing") script_info.clear_group_id(command_script=command_script) - async def set_group_id(self, script_info): + async def set_group_id(self, script_info: ScriptInfo) -> None: """Set or clear the group ID for a script. Parameters @@ -803,7 +829,7 @@ async def set_group_id(self, script_info): except Exception: self.log.exception("next_visit_callback failed; continuing") - def _script_info_from_data(self, event_name, data): + def _script_info_from_data(self, event_name: str, data: BaseMsgType) -> ScriptInfo | None: """Get script info for the script specified in Script event data Parameters @@ -832,17 +858,17 @@ def _script_info_from_data(self, event_name, data): return None return script_info - async def _script_metadata_callback(self, data): + async def _script_metadata_callback(self, data: BaseMsgType) -> None: script_info = self._script_info_from_data(event_name="metadata", data=data) if script_info: script_info.metadata = data - async def _script_state_callback(self, data): + async def _script_state_callback(self, data: BaseMsgType) -> None: script_info = self._script_info_from_data(event_name="state", data=data) if script_info: await script_info._script_state_callback(data) - async def _script_info_callback(self, script_info): + async def _script_info_callback(self, script_info: ScriptInfo) -> None: """ScriptInfo callback.""" self.log.debug(f"Script info callback: {script_info.index}::{script_info.script_state!r}.") if self.script_callback: @@ -861,7 +887,7 @@ async def _script_info_callback(self, script_info): # or be ready to be run. await self._update_queue(force_callback=False) - async def _update_queue(self, force_callback=True, pause_on_failure=True): + async def _update_queue(self, force_callback: bool = True, pause_on_failure: bool = True) -> None: """Call whenever the queue changes state. If the current script is done, move it to the history queue. diff --git a/python/lsst/ts/scriptqueue/run_one_script.py b/python/lsst/ts/scriptqueue/run_one_script.py index 887fe84..5895fb3 100755 --- a/python/lsst/ts/scriptqueue/run_one_script.py +++ b/python/lsst/ts/scriptqueue/run_one_script.py @@ -25,11 +25,16 @@ import asyncio import datetime import logging +import os import pathlib import random +from collections.abc import Sequence +from typing import Any import astropy + from lsst.ts import salobj +from lsst.ts.xml.type_hints import BaseMsgType from .script_queue import SCRIPT_INDEX_MULT, ScriptInfo @@ -39,7 +44,13 @@ class ConfigAction(argparse.Action): """Read config from a file.""" - def __call__(self, parser, namespace, value, option_string=None): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str | Sequence[Any] | None, + option_string: str | None = None, + ) -> None: """Read config from a file. Parameters @@ -54,9 +65,9 @@ def __call__(self, parser, namespace, value, option_string=None): option_string : `str`, optional Option value specified by the user. """ - if not pathlib.Path(value).is_file(): - parser.error(f"Cannot find --config file {value}") - with open(value, "r") as f: + if values is None or not pathlib.Path(str(values)).is_file(): + parser.error(f"Cannot find --config file {values!r}") + with open(str(values), "r") as f: config = f.read() namespace.config = config @@ -64,7 +75,13 @@ def __call__(self, parser, namespace, value, option_string=None): class ParameterAction(argparse.Action): """Parse name=value pairs as config.""" - def __call__(self, parser, namespace, values, option_string): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str | Sequence[Any] | None, + option_string: str | None = None, + ) -> None: """Parse name=value pairs as config. Parameters @@ -79,6 +96,8 @@ def __call__(self, parser, namespace, values, option_string): Option value specified by the user. """ config_list = [] + if values is None: + return for nameValue in values: name, sep, valueStr = nameValue.partition("=") if not valueStr: @@ -87,7 +106,7 @@ def __call__(self, parser, namespace, values, option_string): namespace.config = "\n".join(config_list) -def parse_run_one_script_cmd(args=None): +def parse_run_one_script_cmd(args: list[str] | None = None) -> argparse.Namespace: """Parse command-line arguments for run_one_script.""" description = "Run one SAL script." @@ -129,7 +148,7 @@ def parse_run_one_script_cmd(args=None): return cmd -async def run_one_script(index, script, config, loglevel=None): +async def run_one_script(index: int, script: os.PathLike, config: str, loglevel: int | None = None) -> None: """Run one SAL script. Parameters @@ -160,7 +179,7 @@ async def run_one_script(index, script, config, loglevel=None): ) await remote.start_task - async def log_callback(data): + async def log_callback(data: BaseMsgType) -> None: iso_time = datetime.datetime.now().time().isoformat(timespec="milliseconds") print(f"{iso_time} {logging.getLevelName(data.level)}: {data.message}") @@ -183,7 +202,8 @@ async def log_callback(data): print("waiting for the script to load") await script_info.start_task print("waiting for the script to be configured") - await script_info.config_task + if script_info.config_task is not None: + await script_info.config_task if loglevel is not None: print(f"setting script log level to {loglevel}") await remote.cmd_setLogLevel.set_start(level=loglevel, timeout=STD_TIMEOUT) @@ -192,7 +212,8 @@ async def log_callback(data): await script_info.set_group_id(group_id=group_id) print("running the script") script_info.run() - await script_info.process_task + if script_info.process_task is not None: + await script_info.process_task print("script succeeded") except BaseException: # make sure the background process is terminated @@ -200,12 +221,12 @@ async def log_callback(data): raise -async def _run_one_script_cli_impl(): +async def _run_one_script_cli_impl() -> None: """Implementation for run_one_script_cli.""" cmd = parse_run_one_script_cmd() await run_one_script(index=cmd.index, script=cmd.script, config=cmd.config, loglevel=cmd.loglevel) -def run_one_script_cli(): +def run_one_script_cli() -> None: """Use the command line to run one script.""" asyncio.run(_run_one_script_cli_impl()) diff --git a/python/lsst/ts/scriptqueue/script_info.py b/python/lsst/ts/scriptqueue/script_info.py index 9de0df2..84f29e1 100644 --- a/python/lsst/ts/scriptqueue/script_info.py +++ b/python/lsst/ts/scriptqueue/script_info.py @@ -23,12 +23,18 @@ import asyncio import inspect +import logging import os +from collections.abc import Callable, Coroutine +from typing import Any, Self +from lsst.ts import salobj from lsst.ts.utils import current_tai from lsst.ts.xml.enums.Script import ScriptState from lsst.ts.xml.enums.ScriptQueue import ScriptProcessState +from .type_hints import Indexed, ScriptMetadataProtocol + _CONFIGURE_TIMEOUT = 60 # Time limit for the configure command (seconds) _SET_GROUP_ID_TIMEOUT = 5 # Time limit for setGroupId command (seconds) _TERMINATE_TIMEOUT = 2 # Time limit for terminating script (seconds) @@ -75,19 +81,19 @@ class ScriptInfo: def __init__( self, - log, - remote, - index, - seq_num, - is_standard, - path, - config, - descr, - block="", - log_level=0, - pause_checkpoint="", - stop_checkpoint="", - verbose=False, + log: logging.Logger, + remote: salobj.Remote, + index: int, + seq_num: int, + is_standard: bool, + path: str | os.PathLike, + config: str, + descr: str, + block: str = "", + log_level: int = 0, + pause_checkpoint: str = "", + stop_checkpoint: str = "", + verbose: bool = False, ): self.log = log.getChild(f"ScriptInfo(index={index})") self.remote = remote @@ -107,27 +113,27 @@ def __init__( self.group_id = "" self.verbose = verbose # Most recent value of script metadata; None until set. - self.metadata = None + self.metadata: ScriptMetadataProtocol | None = None # The most recent state reported by the Script, # or 0 if the script is not yet loaded. self.script_state = 0 # Delay between when the script sent state and it was received (sec) self.state_delay = 0 # Time at which the script process was started. 0 before that. - self.timestamp_process_start = 0 + self.timestamp_process_start: float = 0 # Time at which the _configure method was started. 0 before that. # Note: under unusual circumstances the _configure method may fail # before the configure command is sent to the task. - self.timestamp_configure_start = 0 + self.timestamp_configure_start: float = 0 # Time at which the configure command finished (succeeded or failed). # 0 before that. - self.timestamp_configure_end = 0 + self.timestamp_configure_end: float = 0 # Time at which the script started running. 0 before that. - self.timestamp_run_start = 0 + self.timestamp_run_start: float = 0 # Time at which the script process finished. 0 before that. - self.timestamp_process_end = 0 + self.timestamp_process_end: float = 0 # Task for creating self.process, or None if just beginning to load. - self.create_process_task = None + self.create_process_task: asyncio.Task[asyncio.subprocess.Process] | None = None # Task that finishes when configuration starts. # By the time start_task is done config_task and process_task # will exist (instead of being None). Thus: @@ -135,22 +141,22 @@ def __init__( # then await config_task. # * To wait for a script to finish: first await start_task # then await process_task: - self.start_task = asyncio.Future() + self.start_task: asyncio.Future[None] = asyncio.Future() # Process in which the ``Script`` SAL component is loaded. - self.process = None + self.process: asyncio.subprocess.Process | None = None # Task awaiting ``process.wait()``, or None if # the process has not yet started. - self.process_task = None + self.process_task: asyncio.Task[int | None] | None = None # Task awaiting configuration to complete, or None if # configuration has not yet started. - self.config_task = None + self.config_task: asyncio.Future[None] | None = None # Task awaiting clearing group ID. None if group ID not cleared. # Reset to None when group ID is set. - self.clear_group_id_task = None + self.clear_group_id_task: asyncio.Future[None] | None = None # Task awaiting setting group ID, None if group ID is not set. # Reset to None when group ID is cleared. - self.set_group_id_task = None - self._callback = None + self.set_group_id_task: asyncio.Future[None] | None = None + self._callback: Callable[[Self], Coroutine[Any, Any, None]] | None = None # The following guarantees that if we terminate a process # and it sucessfully stops, then we can report it as terminated; @@ -159,7 +165,7 @@ def __init__( self._terminated = False @property - def callback(self): + def callback(self) -> Callable[[Self], Coroutine[Any, Any, None]] | None: """Set, clear or get a callback coroutine (async function) to call whenever the script state changes. @@ -174,40 +180,41 @@ def callback(self): return self._callback @callback.setter - def callback(self, callback): + def callback(self, callback: Callable[[Self], Coroutine[Any, Any, None]] | None) -> None: if callback is not None and not inspect.iscoroutinefunction(callback): raise TypeError(f"callback={callback} must be a coroutine or None") self._callback = callback @property - def configured(self): + def configured(self) -> bool: """True if the configure command succeeded.""" - return self._configure_run and self.config_task.exception() is None + return self.config_task is not None and self._configure_run and self.config_task.exception() is None @property - def configure_failed(self): + def configure_failed(self) -> bool: """True if the configure command failed.""" - return self.script_state == ScriptState.CONFIGURE_FAILED or ( - self._configure_run and self.config_task.exception() is not None + return self.config_task is not None and ( + self.script_state == ScriptState.CONFIGURE_FAILED + or (self._configure_run and self.config_task.exception() is not None) ) @property - def load_failed(self): + def load_failed(self) -> bool: """True if the script could not be loaded.""" return self.process_done and self.timestamp_configure_start == 0 @property - def running(self): + def running(self) -> bool: """True if the script was commanded to run and is not done.""" return self.timestamp_run_start > 0 and not self.process_done @property - def started(self): + def started(self) -> bool: """True if the script was commanded to run or terminate.""" return self.timestamp_run_start > 0 or self.terminated or self.process_done @property - def process_done(self): + def process_done(self) -> bool: """True if the script process was started and is done. Notes @@ -220,16 +227,18 @@ def process_done(self): return self.process_task is not None and self.process_task.done() @property - def failed(self): + def failed(self) -> bool: """True if the script failed. This will be false if the script was terminated.""" if not self.process_done: return False - return self.process.returncode is not None and self.process.returncode > 0 + return ( + self.process is not None and self.process.returncode is not None and self.process.returncode > 0 + ) @property - def terminated(self): + def terminated(self) -> bool: """True if the script was terminated. Notes @@ -244,10 +253,10 @@ def terminated(self): return True if not self.process_done: return False - return self.process.returncode is None or self.process.returncode < 0 + return self.process is None or self.process.returncode is None or self.process.returncode < 0 @property - def process_state(self): + def process_state(self) -> ScriptProcessState: """State of the script subprocess. One of the `ScriptProcessState` enumeration constants. @@ -266,7 +275,7 @@ def process_state(self): return ScriptProcessState.CONFIGURED return ScriptProcessState.LOADING - def run(self): + def run(self) -> None: """Start the script running. Raises @@ -284,21 +293,21 @@ def run(self): self.timestamp_run_start = current_tai() @property - def runnable(self): + def runnable(self) -> bool: """Can the script be run? For a script to be runnable it must be configured, not started, and it must have a group ID. """ - return self.configured and not self.started and self.group_id + return self.configured and not self.started and bool(self.group_id) @property - def setting_group_id(self): + def setting_group_id(self) -> bool: """Return True if the group ID is being set.""" - return self.set_group_id_task and not self.set_group_id_task.done() + return self.set_group_id_task is not None and not self.set_group_id_task.done() @property - def needs_group_id(self): + def needs_group_id(self) -> bool: """Is this script ready to be assigned a group ID? True if the script is configured and not started, @@ -306,7 +315,7 @@ def needs_group_id(self): """ return self.configured and not self.started and not self.group_id and not self.setting_group_id - def clear_group_id(self, command_script): + def clear_group_id(self, command_script: bool) -> None: """Clear the group ID. Can be called in any state. @@ -326,7 +335,7 @@ def clear_group_id(self, command_script): ) ) - def set_block_index(self, block_index): + def set_block_index(self, block_index: int) -> None: """Set the block index for this script. Parameters @@ -336,7 +345,7 @@ def set_block_index(self, block_index): """ self.block_index = int(block_index) - def set_block_id(self, block_id): + def set_block_id(self, block_id: str) -> None: """Set block id, this is unique identifier for a block execution. Parameters @@ -346,7 +355,7 @@ def set_block_id(self, block_id): """ self.block_id = str(block_id) - async def set_group_id(self, group_id): + async def set_group_id(self, group_id: str) -> None: """Set the group ID. Also creates ``self.set_group_id_task`` and sets it done on success @@ -381,7 +390,7 @@ async def set_group_id(self, group_id): self.group_id = group_id await self._run_callback() - async def start_loading(self, fullpath): + async def start_loading(self, fullpath: os.PathLike) -> None: """Start the script process and start a task that will configure the script when it is ready. @@ -432,7 +441,7 @@ async def start_loading(self, fullpath): if not self.timestamp_process_start == 0: await self._run_callback() - async def terminate(self): + async def terminate(self) -> bool: """Terminate the script and wait for the process to terminate. If terminating the script process takes longer than _TERMINATE_TIMEOUT, @@ -470,7 +479,7 @@ async def terminate(self): await self._run_callback() return self._terminated - async def _terminate_process(self): + async def _terminate_process(self) -> None: """Terminate the script process and wait for it to terminate. If necessary, first wait for the process to finish being created. @@ -484,20 +493,26 @@ async def _terminate_process(self): self.process.terminate() await self.process.wait() - def __eq__(self, other): + def __eq__(self, other: object) -> bool: + + if not isinstance(other, Indexed): + return NotImplemented return self.index == other.index - def __ne__(self, other): + def __ne__(self, other: object) -> bool: + + if not isinstance(other, Indexed): + return NotImplemented return not (self == other) - def __repr__(self): + def __repr__(self) -> str: return ( f"ScriptInfo(index={self.index}, seq_num={self.seq_num}, " f"is_standard={self.is_standard}, path={self.path}, " f"config={self.config}, descr={self.descr})" ) - def _cancel_set_clear_group_id(self): + def _cancel_set_clear_group_id(self) -> None: """Cancel set and/or clear group ID tasks, if running. Set the tasks to None. @@ -510,7 +525,7 @@ def _cancel_set_clear_group_id(self): self.set_group_id_task.cancel() self.set_group_id_task = None - def _cleanup(self, returncode=None): + def _cleanup(self, returncode: asyncio.Task[int] | None = None) -> None: """Clean up when the Script subprocess exits. Set the timestamp_process_end, cancel the config task, @@ -522,12 +537,12 @@ def _cleanup(self, returncode=None): self._cancel_set_clear_group_id() asyncio.create_task(self._finish_cleanup()) - async def _finish_cleanup(self): + async def _finish_cleanup(self) -> None: await self._run_callback() self.callback = None self.remote = None - async def _configure(self): + async def _configure(self) -> None: """Configure the script. If configuration fails or is cancelled then terminate the script. @@ -564,15 +579,15 @@ async def _configure(self): self.timestamp_configure_end = current_tai() @property - def _configure_run(self): + def _configure_run(self) -> bool: """Return True if the _configure method was run.""" return self.config_task is not None and self.config_task.done() - async def _run_callback(self, *args): + async def _run_callback(self, *args: Any) -> None: if self.callback: await self.callback(self) - def _trigger_callback(self, *args): + def _trigger_callback(self, *args: Any) -> None: """Synchonous version of _run_callback. Call _run_callback directly, if possible. @@ -580,7 +595,7 @@ def _trigger_callback(self, *args): if self.callback: asyncio.create_task(self._run_callback()) - async def _script_state_callback(self, state): + async def _script_state_callback(self, state: ScriptState) -> None: self.script_state = state.state self.state_delay = current_tai() - state.private_sndStamp if self.script_state == ScriptState.UNCONFIGURED and self.config_task is None: diff --git a/python/lsst/ts/scriptqueue/script_queue.py b/python/lsst/ts/scriptqueue/script_queue.py index 752bfd3..9538368 100644 --- a/python/lsst/ts/scriptqueue/script_queue.py +++ b/python/lsst/ts/scriptqueue/script_queue.py @@ -21,17 +21,25 @@ __all__ = ["ScriptQueue", "run_script_queue"] +import argparse import asyncio import os import subprocess +from pathlib import Path +from typing import Any import numpy as np + from lsst.ts import salobj +from lsst.ts.utils import current_tai from lsst.ts.xml.enums.ScriptQueue import SalIndex +from lsst.ts.xml.sal_enums import State +from lsst.ts.xml.type_hints import BaseMsgType from . import __version__, utils from .queue_model import QueueModel from .script_info import ScriptInfo +from .type_hints import ScriptInfoProtocol SCRIPT_INDEX_MULT = 100000 """Minimum Script SAL index is ScriptQueue SAL index * SCRIPT_INDEX_MULT @@ -75,11 +83,11 @@ class ScriptQueue(salobj.BaseCsc): def __init__( self, - index, - initial_state=salobj.State.STANDBY, - standardpath=None, - externalpath=None, - verbose=False, + index: int, + initial_state: State = State.STANDBY, + standardpath: str | os.PathLike | None = None, + externalpath: str | os.PathLike | None = None, + verbose: bool = False, ): if index < 0 or index > _MAX_SCRIPTQUEUE_INDEX: raise ValueError(f"index {index} must be >= 0 and <= {_MAX_SCRIPTQUEUE_INDEX}") @@ -108,7 +116,7 @@ def __init__( verbose=verbose, ) - def _get_scripts_path(self, patharg, is_standard): + def _get_scripts_path(self, patharg: str | os.PathLike | None, is_standard: bool) -> os.PathLike: """Get the scripts path from the ``standardpath`` or ``externalpath`` constructor argument. @@ -135,25 +143,25 @@ def _get_scripts_path(self, patharg, is_standard): if patharg is None: dir_path = utils.get_default_scripts_dir(is_standard) else: - dir_path = patharg + dir_path = Path(patharg) if not os.path.isdir(dir_path): category = "standard" if is_standard else "external" raise ValueError(f"{category} scripts path {dir_path} is not a directory") - return dir_path + return Path(dir_path) - async def start(self): + async def start(self) -> None: """Finish creating the script queue.""" await super().start() blank_data = self.evt_summaryState.DataType() if hasattr(blank_data, "get_vars"): - def get_data_dict(data): + def get_data_dict(data: BaseMsgType) -> dict[str, Any]: return data.get_vars() else: - def get_data_dict(data): + def get_data_dict(data: BaseMsgType) -> dict[str, Any]: return vars(data) # A function to return message data as a dict of key: value @@ -179,12 +187,12 @@ def get_data_dict(data): ) await self.put_queue() - async def close_tasks(self): + async def close_tasks(self) -> None: """Shut down the queue, terminate all scripts and free resources.""" await self.model.close() await super().close_tasks() - async def do_showAvailableScripts(self, data=None): + async def do_showAvailableScripts(self, data: BaseMsgType | None = None) -> None: """Output a list of available scripts. Parameters @@ -200,7 +208,7 @@ async def do_showAvailableScripts(self, data=None): force_output=True, ) - async def do_showSchema(self, data): + async def do_showSchema(self, data: BaseMsgType) -> None: """Output the config schema for a script. Parameters @@ -232,7 +240,7 @@ async def do_showSchema(self, data): finally: os.environ["PATH"] = initialpath - async def do_showQueue(self, data): + async def do_showQueue(self, data: BaseMsgType) -> None: """Output the queue event. Parameters @@ -243,7 +251,7 @@ async def do_showQueue(self, data): self.assert_enabled("showQueue") await self.put_queue() - async def do_showScript(self, data): + async def do_showScript(self, data: BaseMsgType) -> None: """Output the script event for one script. Parameters @@ -258,7 +266,7 @@ async def do_showScript(self, data): raise salobj.ExpectedError(f"Unknown script {data.scriptSalIndex}") await self.put_script(script_info, force_output=True) - async def do_pause(self, data): + async def do_pause(self, data: BaseMsgType) -> None: """Pause the queue. A no-op if already paused. Unlike most commands, this can be issued in any state. @@ -270,7 +278,7 @@ async def do_pause(self, data): """ await self.model.set_running(False) - async def do_resume(self, data): + async def do_resume(self, data: BaseMsgType) -> None: """Run the queue. A no-op if already running. Parameters @@ -281,7 +289,7 @@ async def do_resume(self, data): self.assert_enabled("resume") await self.model.set_running(True) - async def do_add(self, data): + async def do_add(self, data: BaseMsgType) -> None: """Add a script to the queue. Start and configure a script SAL component, but don't run it. @@ -320,7 +328,7 @@ async def do_add(self, data): result=str(script_info.index), ) - async def do_move(self, data): + async def do_move(self, data: BaseMsgType) -> None: """Move a script within the queue.""" self.assert_enabled("move") try: @@ -332,7 +340,7 @@ async def do_move(self, data): except ValueError as e: raise salobj.ExpectedError(str(e)) - async def do_requeue(self, data): + async def do_requeue(self, data: BaseMsgType) -> None: """Put a script back on the queue with the same configuration.""" self.assert_enabled("requeue") try: @@ -345,7 +353,7 @@ async def do_requeue(self, data): except ValueError as e: raise salobj.ExpectedError(str(e)) - async def do_stopScripts(self, data): + async def do_stopScripts(self, data: BaseMsgType) -> None: """Stop one or more queued scripts and/or the current script. If you stop the current script, it is moved to the history. @@ -361,14 +369,14 @@ async def do_stopScripts(self, data): timeout, ) - async def handle_summary_state(self): + async def handle_summary_state(self) -> None: await super().handle_summary_state() enabled = self.summary_state == salobj.State.ENABLED await self.model.set_enable(enabled) if enabled: await self.do_showAvailableScripts() - async def put_next_visit(self, script_info): + async def put_next_visit(self, script_info: ScriptInfoProtocol) -> None: """Output the ``nextVisit`` event.""" if self.verbose: print(f"put_next_visit: index={script_info.index}, group_id={script_info.group_id}") @@ -381,14 +389,33 @@ async def put_next_visit(self, script_info): for key, value in self.get_data_dict(script_info.metadata).items() if key not in self.base_field_names } + + if self.model.current_script is None or self.model.current_script == script_info: + next_visit_start_time = current_tai() + else: + duration = ( + self.model.current_script.metadata.duration + if self.model.current_script.metadata is not None + else 0.0 + ) + next_visit_start_time = self.model.current_script.timestamp_run_start + duration + next_visit_start_time_minimum = current_tai() + + next_visit_start_time = ( + next_visit_start_time + if next_visit_start_time > next_visit_start_time_minimum + else next_visit_start_time_minimum + ) + await self.evt_nextVisit.set_write( scriptSalIndex=script_info.index, groupId=script_info.group_id, + startTime=next_visit_start_time, **metadata_dict, force_output=True, ) - async def put_next_visit_canceled(self, script_info): + async def put_next_visit_canceled(self, script_info: ScriptInfoProtocol) -> None: """Output the ``nextVisitCanceled`` event.""" if self.verbose: print(f"put_next_visit_canceled: index={script_info.index}, group_id={script_info.group_id}") @@ -400,7 +427,7 @@ async def put_next_visit_canceled(self, script_info): force_output=True, ) - async def put_queue(self): + async def put_queue(self) -> None: """Output the queued scripts as a ``queue`` event. The data is put even if the queue has not changed. That way commands @@ -435,7 +462,7 @@ async def put_queue(self): force_output=True, ) - async def put_script(self, script_info, force_output=False): + async def put_script(self, script_info: ScriptInfoProtocol, force_output: bool = False) -> None: """Output information about a script as a ``script`` event. Designed to be used as a QueueModel script_callback. @@ -456,6 +483,7 @@ async def put_script(self, script_info, force_output=False): f"process_state={script_info.process_state}, " f"script_state={script_info.script_state}" ) + await self.evt_script.set_write( cmdId=script_info.seq_num, scriptSalIndex=script_info.index, @@ -472,7 +500,7 @@ async def put_script(self, script_info, force_output=False): ) @classmethod - def add_arguments(cls, parser): + def add_arguments(cls, parser: argparse.ArgumentParser) -> None: parser.add_argument( "--standard", help="Directory containing standard scripts; defaults to ts_standardscripts/scripts", @@ -488,12 +516,12 @@ def add_arguments(cls, parser): ) @classmethod - def add_kwargs_from_args(cls, args, kwargs): + def add_kwargs_from_args(cls, args: argparse.Namespace, kwargs: dict[str, Any]) -> None: kwargs["standardpath"] = args.standard kwargs["externalpath"] = args.external kwargs["verbose"] = args.verbose -def run_script_queue(): +def run_script_queue() -> None: """Run the ScriptQueue CSC.""" asyncio.run(ScriptQueue.amain(index=SalIndex)) diff --git a/python/lsst/ts/scriptqueue/script_queue_commander.py b/python/lsst/ts/scriptqueue/script_queue_commander.py index 666a30a..dab7c17 100644 --- a/python/lsst/ts/scriptqueue/script_queue_commander.py +++ b/python/lsst/ts/scriptqueue/script_queue_commander.py @@ -21,15 +21,18 @@ __all__ = ["ScriptQueueCommander"] +import argparse import asyncio import logging import pathlib import string +from typing import Any from lsst.ts import salobj from lsst.ts.utils import make_done_future from lsst.ts.xml.enums.Script import ScriptState from lsst.ts.xml.enums.ScriptQueue import Location, SalIndex +from lsst.ts.xml.type_hints import BaseMsgType ADD_TIMEOUT = 5 # Timeout for the add command (seconds). # How long to wait before warning that a script heartbeat is late (seconds). @@ -45,7 +48,7 @@ class ScriptQueueCommander(salobj.CscCommander): Default log level for scripts. """ - def __init__(self, script_log_level, **kwargs): + def __init__(self, script_log_level: int, **kwargs: Any) -> None: super().__init__(name="ScriptQueue", **kwargs) self.script_log_level = script_log_level self.help_dict["add"] = f"""type path config options # add a script to the end of the queue: @@ -92,18 +95,18 @@ def __init__(self, script_log_level, **kwargs): self._script_to_monitor = 0 self.script_heartbeat_monitor_task = make_done_future() - async def start(self): + async def start(self) -> None: await super().start() await self.script_remote.start_task - def get_is_standard(self, script_type): + def get_is_standard(self, script_type: str) -> bool: """Convert a script type argument to isStandard bool.""" try: return self.script_type_dict[script_type] except KeyError: raise KeyError(f"type {script_type!r} must be one of {list(self.script_type_dict.keys())}") - def evt_availableScripts_callback(self, data): + def evt_availableScripts_callback(self, data: BaseMsgType) -> None: standard_scripts = data.standard.split(":") external_scripts = data.external.split(":") print("standard scripts:") @@ -113,7 +116,7 @@ def evt_availableScripts_callback(self, data): for name in external_scripts: print(f"• {name}") - def evt_queue_callback(self, data): + def evt_queue_callback(self, data: BaseMsgType) -> None: if self._script_to_monitor != data.currentSalIndex: self._script_to_monitor = data.currentSalIndex self.script_heartbeat_monitor_task.cancel() @@ -130,7 +133,7 @@ def evt_queue_callback(self, data): f"pastSalIndices={pastSalIndices}" ) - async def script_heartbeat_monitor(self): + async def script_heartbeat_monitor(self) -> None: while True: await asyncio.sleep(HEARTBEAT_ALARM_INTERVAL) print( @@ -138,7 +141,7 @@ async def script_heartbeat_monitor(self): f"heartbeat not seen in {HEARTBEAT_ALARM_INTERVAL} seconds" ) - async def script_log_message(self, data): + async def script_log_message(self, data: BaseMsgType) -> None: exception_str = ( ( f", traceback={data.traceback}, " @@ -155,7 +158,7 @@ async def script_log_message(self, data): f"message={data.message}{exception_str}" ) - async def script_state(self, data): + async def script_state(self, data: BaseMsgType) -> None: try: state = ScriptState(data.state) except ValueError: @@ -166,14 +169,14 @@ async def script_state(self, data): f"state={state.name}{reason}, lastCheckpoint={data.lastCheckpoint}" ) - async def script_heartbeat(self, data): + async def script_heartbeat(self, data: BaseMsgType) -> None: if data.salIndex != self._script_to_monitor: # A heartbeat from the wrong script. return self.script_heartbeat_monitor_task.cancel() self.script_heartbeat_monitor_task = asyncio.create_task(self.script_heartbeat_monitor()) - async def do_add(self, args): + async def do_add(self, args: list[Any]) -> None: """Overrride the standard add command to simplify the interface.""" if len(args) < 2: raise ValueError("Need at least 2 arguments") @@ -237,7 +240,7 @@ async def do_add(self, args): timeout=ADD_TIMEOUT, ) - async def do_showSchema(self, args): + async def do_showSchema(self, args: list[str]) -> None: """Overrride the standard showSchema command for named script type.""" if len(args) != 2: raise ValueError("Need 2 arguments: type path") @@ -248,7 +251,7 @@ async def do_showSchema(self, args): path=path, ) - async def do_stopScripts(self, args): + async def do_stopScripts(self, args: list[str]) -> None: """Handle the stopScript command, which takes a list of script indices. """ @@ -267,7 +270,7 @@ async def do_stopScripts(self, args): await self.remote.cmd_stopScripts.start(data=stop_data) @classmethod - def add_arguments(cls, parser): + def add_arguments(cls, parser: argparse.ArgumentParser) -> None: parser.add_argument( "-l", "--loglevel", @@ -277,11 +280,11 @@ def add_arguments(cls, parser): ) @classmethod - def add_kwargs_from_args(cls, args, kwargs): + def add_kwargs_from_args(cls, args: argparse.Namespace, kwargs: dict[str, Any]) -> None: kwargs["script_log_level"] = args.loglevel -def command_script_queue(): +def command_script_queue() -> None: """Run a command-line interface to command a ScriptQueue. Intended for engineering use. diff --git a/python/lsst/ts/scriptqueue/type_hints.py b/python/lsst/ts/scriptqueue/type_hints.py new file mode 100644 index 0000000..046bd25 --- /dev/null +++ b/python/lsst/ts/scriptqueue/type_hints.py @@ -0,0 +1,70 @@ +# This file is part of ts_scriptqueue. +# +# Developed for the LSST Telescope and Site Systems. +# This product includes software developed by the LSST Project +# (https://www.lsst.org). +# See the COPYRIGHT file at the top-level directory of this distribution +# for details of code ownership. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +from collections.abc import Awaitable, Callable, Coroutine +from typing import Any, Protocol, runtime_checkable + +from lsst.ts.xml.enums.ScriptQueue import ScriptProcessState + + +@runtime_checkable +class Indexed(Protocol): + index: int + + +class ScriptMetadataProtocol(Protocol): + duration: float + + +class ScriptInfoProtocol(Protocol): + index: int + group_id: str + metadata: ScriptMetadataProtocol | None + script_state: int + seq_num: int + path: str + is_standard: bool + timestamp_process_start: float + timestamp_configure_start: float + timestamp_configure_end: float + timestamp_run_start: float + timestamp_process_end: float + + def set_block_id(self, block_id: str) -> None: ... + + def set_block_index(self, block_index: int) -> None: ... + + @property + def process_done(self) -> bool: ... + + @property + def process_state(self) -> ScriptProcessState: ... + + @property + def running(self) -> bool: ... + + +class AsyncScriptInfoBoolCallback(Protocol): + def __call__(self, script_info: ScriptInfoProtocol, force_output: bool = False) -> Awaitable[None]: ... + + +AsyncScriptInfoCallback = Callable[[ScriptInfoProtocol], Coroutine[Any, Any, None]] +AsyncNoArgsCallback = Callable[[], Coroutine[Any, Any, None]] diff --git a/python/lsst/ts/scriptqueue/utils.py b/python/lsst/ts/scriptqueue/utils.py index fed6ded..837dba8 100644 --- a/python/lsst/ts/scriptqueue/utils.py +++ b/python/lsst/ts/scriptqueue/utils.py @@ -31,7 +31,7 @@ import time -def find_public_scripts(root): +def find_public_scripts(root: str) -> list[str]: """Find all public scripts in the specified root path. Public scripts are executable files whose names do not start @@ -55,7 +55,9 @@ def find_public_scripts(root): return [os.path.relpath(exe, root) for exe in executables] -def configure_logging(verbose=0, console_format=None, filename=None): +def configure_logging( + verbose: int = 0, console_format: str | None = None, filename: str | None = None +) -> None: """Configure the logging for the system. Parameters @@ -89,13 +91,14 @@ def configure_logging(verbose=0, console_format=None, filename=None): ch.setFormatter(logging.Formatter(console_format)) logging.getLogger().addHandler(ch) - log_file = logging.FileHandler(filename) - log_file.setFormatter(logging.Formatter(log_format)) - log_file.setLevel(file_detail) - logging.getLogger().addHandler(log_file) + if filename is not None: + log_file = logging.FileHandler(filename) + log_file.setFormatter(logging.Formatter(log_format)) + log_file.setLevel(file_detail) + logging.getLogger().addHandler(log_file) -def generate_logfile(basename="scriptqueue"): +def generate_logfile(basename: str = "scriptqueue") -> str: """Generate a log file name based on current time.""" timestr = time.strftime("%Y-%m-%d_%H:%M:%S") log_path = os.path.expanduser("~/.{}/log".format(basename)) @@ -105,7 +108,7 @@ def generate_logfile(basename="scriptqueue"): return logfilename -def get_default_scripts_dir(is_standard): +def get_default_scripts_dir(is_standard: bool) -> os.PathLike: """Return the default directory for the specified kind of scripts. Parameters diff --git a/tests/data/external/script1 b/tests/data/external/script7 similarity index 100% rename from tests/data/external/script1 rename to tests/data/external/script7 diff --git a/tests/data/external/subdir/script3 b/tests/data/external/subdir/script8 similarity index 100% rename from tests/data/external/subdir/script3 rename to tests/data/external/subdir/script8 diff --git a/tests/test_queue_model.py b/tests/test_queue_model.py index e7e93b2..194e926 100644 --- a/tests/test_queue_model.py +++ b/tests/test_queue_model.py @@ -26,9 +26,12 @@ import time import unittest import warnings -from unittest.mock import patch +from collections.abc import Iterable, Sequence +from typing import Any, Generator +from unittest.mock import Mock, patch import pytest + from lsst.ts import salobj, scriptqueue from lsst.ts.xml import subsystems from lsst.ts.xml.enums.Script import ScriptState @@ -39,7 +42,7 @@ STD_TIMEOUT = 60 -def _min_sal_index_generator(): +def _min_sal_index_generator() -> Generator[int, None, None]: min_sal_index = 1000 while True: yield min_sal_index @@ -52,7 +55,7 @@ def _min_sal_index_generator(): class QueueInfo: """Information about the queue. Used by assert_next_queue.""" - def __init__(self, model): + def __init__(self, model: scriptqueue.QueueModel) -> None: self.enabled = model.enabled self.running = model.running self.current_index = model.current_index @@ -61,7 +64,7 @@ def __init__(self, model): class QueueModelTestCase(unittest.IsolatedAsyncioTestCase): - async def asyncSetUp(self): + async def asyncSetUp(self) -> None: self.t0 = time.monotonic() self.min_sal_index = next(make_min_sal_index) salobj.set_test_topic_subname() @@ -72,13 +75,13 @@ async def asyncSetUp(self): self.log = logging.getLogger() # Queue of (sal_index, group_id) set by next_visit_callback # and used by assert_next_next_visit - self.next_visit_queue = asyncio.Queue() + self.next_visit_queue: asyncio.Queue[tuple[int, str]] = asyncio.Queue() # Queue of (sal_index, group_id) set by next_visit_canceled_callback # and used by assert_next_next_visit_canceled - self.next_visit_canceled_queue = asyncio.Queue() + self.next_visit_canceled_queue: asyncio.Queue[tuple[int, str]] = asyncio.Queue() # Queue of script queue information; # used by assert_next_queue - self.queue_info_queue = asyncio.Queue() + self.queue_info_queue: asyncio.Queue[QueueInfo] = asyncio.Queue() self.model = scriptqueue.QueueModel( domain=self.domain, log=self.log, @@ -94,7 +97,7 @@ async def asyncSetUp(self): await self.model.set_enable(True) await self.model.start_task - async def asyncTearDown(self): + async def asyncTearDown(self) -> None: killed_scripts_info = await asyncio.wait_for(self.model.terminate_all(), timeout=STD_TIMEOUT) if killed_scripts_info: killed_scripts_index = ",".join([f"{script_info.index}" for script_info in killed_scripts_info]) @@ -124,7 +127,7 @@ async def asyncTearDown(self): # Sleep some time to let the cluster have time to finish the deletion await asyncio.sleep(5.0) - async def assert_next_next_visit(self, sal_index): + async def assert_next_next_visit(self, sal_index: int) -> None: """Assert that the next next_visit callback is for the specified index. Parameters @@ -138,7 +141,7 @@ async def assert_next_next_visit(self, sal_index): assert next_sal_index == sal_index assert next_group_id != "" - async def assert_next_next_visit_canceled(self, sal_index): + async def assert_next_next_visit_canceled(self, sal_index: int) -> None: """Assert that the next next_visit_canceled callback is for the specified index. @@ -155,13 +158,13 @@ async def assert_next_next_visit_canceled(self, sal_index): async def assert_next_queue( self, - enabled=True, - running=False, - current_sal_index=0, - sal_indices=(), - past_sal_indices=(), - wait=True, - ): + enabled: bool = True, + running: bool = False, + current_sal_index: int = 0, + sal_indices: Sequence[int] = (), + past_sal_indices: Iterable[int] = (), + wait: bool = True, + ) -> QueueInfo: """Check next or current queue state. The defaults are appropriate to an enabled, paused queue @@ -228,7 +231,9 @@ async def assert_next_queue( assert actual_past_sal_indices == list(past_sal_indices) return queue_info - def assert_script_info_equal(self, info1, info2, is_requeue=False): + def assert_script_info_equal( + self, info1: scriptqueue.ScriptInfo, info2: scriptqueue.ScriptInfo, is_requeue: bool = False + ) -> None: """Assert two ScriptInfo are equal. If is_requeue (indicating that we are comparing a requeued @@ -248,12 +253,12 @@ def assert_script_info_equal(self, info1, info2, is_requeue=False): def make_add_kwargs( self, - location=Location.LAST, - location_sal_index=0, - is_standard=False, - path=None, - config="wait_time: 0.1", - ): + location: Location = Location.LAST, + location_sal_index: int = 0, + is_standard: bool = False, + path: str | None = None, + config: str = "wait_time: 0.1", + ) -> dict[str, Any]: """Make keyword arguments for QueueModel.add. Parameters @@ -278,7 +283,9 @@ def make_add_kwargs( location_sal_index=location_sal_index, ) - def make_script_info(self, is_standard=False, path=None, config="wait_time: 0.1"): + def make_script_info( + self, is_standard: bool = False, path: str | None = None, config: str = "wait_time: 0.1" + ) -> scriptqueue.ScriptInfo: """Make a `ScriptInfo`. Parameters @@ -304,7 +311,7 @@ def make_script_info(self, is_standard=False, path=None, config="wait_time: 0.1" verbose=True, ) - async def next_visit_callback(self, script_info): + async def next_visit_callback(self, script_info: scriptqueue.ScriptInfo) -> None: dt = time.monotonic() - self.t0 print( f"next_visit_callback() for {script_info.index}: " @@ -313,7 +320,7 @@ async def next_visit_callback(self, script_info): ) await self.next_visit_queue.put((script_info.index, script_info.group_id)) - async def next_visit_canceled_callback(self, script_info): + async def next_visit_canceled_callback(self, script_info: scriptqueue.ScriptInfo) -> None: dt = time.monotonic() - self.t0 print( f"next_visit_canceled_callback() for {script_info.index}: " @@ -322,7 +329,7 @@ async def next_visit_canceled_callback(self, script_info): ) await self.next_visit_canceled_queue.put((script_info.index, script_info.group_id)) - async def queue_callback(self): + async def queue_callback(self) -> None: dt = time.monotonic() - self.t0 print( f"queue_callback(): enabled={self.model.enabled}; " @@ -334,7 +341,7 @@ async def queue_callback(self): ) await self.queue_info_queue.put(QueueInfo(self.model)) - async def script_callback(self, script_info): + async def script_callback(self, script_info: scriptqueue.ScriptInfo) -> None: curr_time = time.monotonic() dt = curr_time - self.t0 print( @@ -349,7 +356,7 @@ async def script_callback(self, script_info): f"state_delay={script_info.state_delay:0.1f}" ) - async def test_add_scripts(self): + async def test_add_scripts(self) -> None: """Test add.""" await self.assert_next_queue(enabled=True, running=True) @@ -429,7 +436,7 @@ async def test_add_scripts(self): # and use it in the remaining tests. queue_info = await self.assert_next_queue( sal_indices=[i0 + 2, i0 + 1, i0 + 3], - past_sal_indices={i0 + 6, i0 + 5, i0, i0 + 4}, + past_sal_indices={int(i0 + 6), int(i0 + 5), int(i0), int(i0 + 4)}, ) stopped_scripts = [info.index for info in queue_info.history] @@ -487,7 +494,7 @@ async def test_add_scripts(self): # Make sure that next_visit_canceled_callback was not called assert self.next_visit_canceled_queue.empty() - async def test_add_blocks_fails_no_image_server_url(self): + async def test_add_blocks_fails_no_image_server_url(self) -> None: """Test adding scripts that are part of a block.""" await self.assert_next_queue(enabled=True, running=True) @@ -507,7 +514,7 @@ async def test_add_blocks_fails_no_image_server_url(self): @patch("lsst.ts.utils.ImageNameServiceClient.get_next_obs_id") @patch.dict(os.environ, {"IMAGE_SERVER_URL": "mytemp"}) - async def test_add_blocks(self, mock_get): + async def test_add_blocks(self, mock_get: Mock) -> None: """Test adding scripts that are part of a block.""" mock_get.side_effect = [ (0, ["BL1_O_20240228_000001"]), @@ -578,7 +585,7 @@ async def test_add_blocks(self, mock_get): @patch("lsst.ts.utils.ImageNameServiceClient.get_next_obs_id") @patch.dict(os.environ, {"IMAGE_SERVER_URL": "mytemp"}) - async def test_add_blocks_more_scripts(self, mock_get): + async def test_add_blocks_more_scripts(self, mock_get: Mock) -> None: """Test adding scripts that are part of a block.""" mock_get.return_value = (0, ["BL1_O_20240228_000001"]) await self.assert_next_queue(enabled=True, running=True) @@ -623,7 +630,7 @@ async def test_add_blocks_more_scripts(self, mock_get): ): await asyncio.wait_for(self.model.add(**add_kwargs), timeout=STD_TIMEOUT) - async def test_add_bad_config(self): + async def test_add_bad_config(self) -> None: """Test adding a script with invalid configuration.""" await self.assert_next_queue(enabled=True, running=True) @@ -647,7 +654,7 @@ async def test_add_bad_config(self): assert script0.process_done assert script0.process_state == ScriptProcessState.CONFIGURE_FAILED - async def check_add_then_stop_script(self, terminate): + async def check_add_then_stop_script(self, terminate: bool) -> None: """Test adding a script immediately followed by stoppping it.""" await self.assert_next_queue(enabled=True, running=True) @@ -671,13 +678,13 @@ async def check_add_then_stop_script(self, terminate): assert not (script0.configured) assert script0.process_state == ScriptProcessState.TERMINATED - async def test_add_then_stop_script(self): + async def test_add_then_stop_script(self) -> None: await self.check_add_then_stop_script(terminate=False) - async def test_add_then_terminate_script(self): + async def test_add_then_terminate_script(self) -> None: await self.check_add_then_stop_script(terminate=True) - def test_constructor_errors(self): + def test_constructor_errors(self) -> None: nonexistentpath = os.path.join(self.datadir, "garbage") with pytest.raises(ValueError): scriptqueue.QueueModel( @@ -701,7 +708,7 @@ def test_constructor_errors(self): externalpath=nonexistentpath, ) - async def test_get_script_info(self): + async def test_get_script_info(self) -> None: await self.assert_next_queue(enabled=True, running=True) # Pause the queue so we know what to expect of queue state. @@ -770,7 +777,7 @@ async def test_get_script_info(self): past_sal_indices=[i0 + 2, i0 + 1, i0], ) - def test_make_full_path(self): + def test_make_full_path(self) -> None: for is_standard, badpath in ( (True, "../script5"), # file is in external, not standard (True, "subdir/nonex2"), # file is not executable @@ -785,7 +792,7 @@ def test_make_full_path(self): for is_standard, goodpath in ( (True, "subdir/subsubdir/script4"), - (False, "subdir/script3"), + (False, "subdir/script8"), (True, "script2"), ): with self.subTest(is_standard=is_standard, path=goodpath): @@ -794,7 +801,7 @@ def test_make_full_path(self): expected_fullpath = os.path.join(root, goodpath) assert fullpath.samefile(expected_fullpath) - async def test_move(self): + async def test_move(self) -> None: """Test move, pause and showQueue""" await self.assert_next_queue(enabled=True, running=True) @@ -901,7 +908,7 @@ async def test_move(self): timeout=STD_TIMEOUT, ) - async def test_clear_group_id(self): + async def test_clear_group_id(self) -> None: """Test that a script at the top of the queue has its group ID cleared if it is moved elsewhere. """ @@ -967,7 +974,7 @@ async def test_clear_group_id(self): past_sal_indices=[i0 + 1, i0 + 2, i0], ) - async def test_pause_on_failure(self): + async def test_pause_on_failure(self) -> None: """Test that a failed script pauses the queue.""" await self.assert_next_queue(enabled=True, running=True) @@ -1048,7 +1055,7 @@ async def test_pause_on_failure(self): assert script_info.process.returncode == 0 assert script_info.script_state == ScriptState.DONE - async def test_requeue(self): + async def test_requeue(self) -> None: """Test requeue""" await self.assert_next_queue(enabled=True, running=True) @@ -1199,7 +1206,7 @@ async def test_requeue(self): for requeue_info, info in zip(requeue_info_list, info_list): self.assert_script_info_equal(requeue_info, info, is_requeue=True) - async def test_resume_before_first_script_runnable(self): + async def test_resume_before_first_script_runnable(self) -> None: await self.assert_next_queue(enabled=True, running=True) # pause the queue so we know what to expect of queue state @@ -1227,7 +1234,7 @@ async def test_resume_before_first_script_runnable(self): ) await self.assert_next_queue(running=True, current_sal_index=0, sal_indices=[], past_sal_indices=[i0]) - async def test_run_immediately(self): + async def test_run_immediately(self) -> None: await self.assert_next_queue(enabled=True, running=True) info0 = self.make_script_info(is_standard=False, path=os.path.join("subdir", "script6"), config="") @@ -1245,7 +1252,7 @@ async def test_run_immediately(self): ) await self.assert_next_queue(running=True, current_sal_index=0, sal_indices=[], past_sal_indices=[i0]) - async def check_stop_scripts(self, terminate): + async def check_stop_scripts(self, terminate: bool) -> None: await self.assert_next_queue(enabled=True, running=True) # pause the queue so we know what to expect of queue state @@ -1391,13 +1398,13 @@ async def check_stop_scripts(self, terminate): timeout=STD_TIMEOUT, ) - async def test_stop_scripts_noterminate(self): + async def test_stop_scripts_noterminate(self) -> None: await self.check_stop_scripts(terminate=False) - async def test_stop_scripts_terminate(self): + async def test_stop_scripts_terminate(self) -> None: await self.check_stop_scripts(terminate=True) - async def wait_done(self, *indices): + async def wait_done(self, *indices: int) -> list[Any]: """Wait for the specified scripts finish running (succeed or fail). Return the result of each task. @@ -1416,7 +1423,7 @@ async def wait_done(self, *indices): late_scripts = [ind for task, ind in zip(process_tasks, indices) if not task.done()] raise RuntimeError(f"Scripts {late_scripts} did not finish in 60 seconds") - async def wait_configured(self, *indices): + async def wait_configured(self, *indices: int) -> None: """Wait for the specified scripts to be configured. Call this before running the queue if you want the queue data @@ -1444,7 +1451,7 @@ async def wait_configured(self, *indices): f"elapsed time={dt:0.1f}" ) from e - async def wait_running(self, sal_index): + async def wait_running(self, sal_index: int) -> None: """Wait for the specified script to report that it is running. Parameters diff --git a/tests/test_run_one_script.py b/tests/test_run_one_script.py index cf6580a..850b4e6 100644 --- a/tests/test_run_one_script.py +++ b/tests/test_run_one_script.py @@ -27,8 +27,10 @@ import pytest import yaml + from lsst.ts import salobj, scriptqueue from lsst.ts.xml.enums.Script import ScriptState +from lsst.ts.xml.type_hints import BaseMsgType # Long enough to perform any reasonable operation # including starting a CSC or loading a script (seconds) @@ -38,14 +40,14 @@ class ParseRunOneScriptTestCase(unittest.IsolatedAsyncioTestCase): - def test_basics(self): + def test_basics(self) -> None: script = DATA_DIR / "standard" / "subdir" / "script3" cmd = scriptqueue.parse_run_one_script_cmd(args=[str(script)]) assert script.samefile(cmd.script) assert cmd.config == "" - def test_config_arg(self): - script = DATA_DIR / "external" / "script1" + def test_config_arg(self) -> None: + script = DATA_DIR / "external" / "script7" config_path = DATA_DIR / "config1.yaml" with open(config_path, "r") as f: expected_config = f.read() @@ -53,8 +55,8 @@ def test_config_arg(self): assert script.samefile(cmd.script) assert cmd.config == expected_config - def test_parameters_arg(self): - script = DATA_DIR / "external" / "script1" + def test_parameters_arg(self) -> None: + script = DATA_DIR / "external" / "script7" config_dict = dict(abool=True, anint=47, afloat=0.2, astr="string_value") config_arg_list = [f"{key}={value}" for key, value in config_dict.items()] cmd = scriptqueue.parse_run_one_script_cmd(args=[str(script), "--parameters"] + config_arg_list) @@ -62,8 +64,8 @@ def test_parameters_arg(self): config_dict_from_parser = yaml.safe_load(cmd.config) assert config_dict_from_parser == config_dict - def test_loglevel(self): - script = DATA_DIR / "external" / "script1" + def test_loglevel(self) -> None: + script = DATA_DIR / "external" / "script7" cmd = scriptqueue.parse_run_one_script_cmd(args=[str(script)]) assert cmd.loglevel is None @@ -75,8 +77,8 @@ def test_loglevel(self): cmd = scriptqueue.parse_run_one_script_cmd(args=[str(script), "-l", str(loglevel)]) assert cmd.loglevel == loglevel - def test_invalid_arguments(self): - script = DATA_DIR / "external" / "script1" + def test_invalid_arguments(self) -> None: + script = DATA_DIR / "external" / "script7" config_path = DATA_DIR / "config1.yaml" with pytest.raises(SystemExit): @@ -129,17 +131,17 @@ def test_invalid_arguments(self): class RunOneScriptTestCase(unittest.IsolatedAsyncioTestCase): - def setUp(self): - salobj.set_random_lsst_dds_partition_prefix() + def setUp(self) -> None: + salobj.set_test_topic_subname() - async def test_run_one_script(self): + async def test_run_one_script(self) -> None: script = DATA_DIR / "standard" / "subdir" / "script3" config_path = DATA_DIR / "config1.yaml" with open(config_path, "r") as f: config = f.read() await scriptqueue.run_one_script(index=1, script=script, config=config, loglevel=10) - async def test_run_command_line(self): + async def test_run_command_line(self) -> None: exe_name = "run_one_script" exe_path = shutil.which(exe_name) if exe_path is None: @@ -154,9 +156,9 @@ async def test_run_command_line(self): ): # The script states seen, ignoring sequential duplicates # (e.g. [1, 1, 2, 2, 1, 1] becomes [1, 2, 1] - states_seen = [] + states_seen: list[ScriptState] = [] - async def state_callback(data): + async def state_callback(data: BaseMsgType) -> None: nonlocal states_seen state = ScriptState(data.state) if not states_seen or state != states_seen[-1]: diff --git a/tests/test_script_queue.py b/tests/test_script_queue.py index 80f2e1f..247099f 100644 --- a/tests/test_script_queue.py +++ b/tests/test_script_queue.py @@ -25,14 +25,18 @@ import os import shutil import unittest -from unittest.mock import patch +from typing import Any, Iterable, Sequence +from unittest.mock import Mock, patch import pytest import yaml + from lsst.ts import salobj, scriptqueue, utils from lsst.ts.xml import subsystems from lsst.ts.xml.enums.Script import ScriptState from lsst.ts.xml.enums.ScriptQueue import Location, SalIndex, ScriptProcessState +from lsst.ts.xml.sal_enums import State +from lsst.ts.xml.type_hints import BaseMsgType try: from lsst.ts import standardscripts @@ -58,10 +62,10 @@ class MakeKWargs: Call the functor with optional overrides. """ - def __init__(self, **defaults): + def __init__(self, **defaults: Any) -> None: self.defaults = defaults - def __call__(self, **kwargs): + def __call__(self, **kwargs: Any) -> Any: ret = copy.copy(self.defaults) ret.update(kwargs) return ret @@ -85,11 +89,11 @@ class MakeAddKwargs(MakeKWargs): def __init__( self, - isStandard="True", - path="script1", - config="wait_time: 0.1", - descr="a description", - ): + isStandard: bool = True, + path: str = "script1", + config: str = "wait_time: 0.1", + descr: str = "a description", + ) -> None: super().__init__( isStandard=isStandard, path=path, @@ -101,7 +105,7 @@ def __init__( class ScriptQueueConstructorTestCase(unittest.IsolatedAsyncioTestCase): - def setUp(self): + def setUp(self) -> None: salobj.set_test_topic_subname() try: self.default_standardpath = scriptqueue.get_default_scripts_dir(is_standard=True) @@ -118,7 +122,7 @@ def setUp(self): self.testdata_externalpath = os.path.join(self.datadir, "external") self.badpath = os.path.join(self.datadir, "not_a_directory") - async def asyncTearDown(self): + async def asyncTearDown(self) -> None: topic_subname = os.environ["LSST_TOPIC_SUBNAME"] delete_topics = await salobj.delete_topics.DeleteTopics.new() @@ -144,7 +148,7 @@ async def asyncTearDown(self): standardscripts is None or externalscripts is None, "Could not import ts_standardscripts and/or ts_externalscripts.", ) - async def test_default_paths(self): + async def test_default_paths(self) -> None: async with ( scriptqueue.ScriptQueue(index=SalIndex.MAIN_TEL) as queue, salobj.Remote(domain=queue.domain, name="ScriptQueue", index=SalIndex.MAIN_TEL) as remote, @@ -159,7 +163,7 @@ async def test_default_paths(self): assert self.testdata_standardpath != self.default_standardpath assert self.testdata_externalpath != self.default_externalpath - async def test_explicit_paths(self): + async def test_explicit_paths(self) -> None: async with ( scriptqueue.ScriptQueue( index=SalIndex.MAIN_TEL, @@ -178,7 +182,7 @@ async def test_explicit_paths(self): standardscripts is None, "Could not import ts_standardscripts.", ) - async def test_default_standard_path(self): + async def test_default_standard_path(self) -> None: async with ( scriptqueue.ScriptQueue( index=SalIndex.MAIN_TEL, externalpath=self.testdata_externalpath @@ -195,7 +199,7 @@ async def test_default_standard_path(self): externalscripts is None, "Could not import ts_externalscripts.", ) - async def test_default_external_path(self): + async def test_default_external_path(self) -> None: async with ( scriptqueue.ScriptQueue( index=SalIndex.MAIN_TEL, standardpath=self.testdata_standardpath @@ -208,7 +212,7 @@ async def test_default_external_path(self): assert os.path.samefile(rootDir_data.standard, self.testdata_standardpath) assert os.path.samefile(rootDir_data.external, self.default_externalpath) - def test_invalid_paths(self): + def test_invalid_paths(self) -> None: with pytest.raises(ValueError): scriptqueue.ScriptQueue( index=SalIndex.MAIN_TEL, @@ -230,14 +234,16 @@ def test_invalid_paths(self): class ScriptQueueTestCase(salobj.BaseCscTestCase, unittest.IsolatedAsyncioTestCase): - def setUp(self): + def setUp(self) -> None: super().setUp() datadir = os.path.abspath(os.path.join(os.path.dirname(__file__), "data")) self.standardpath = os.path.join(datadir, "standard") self.externalpath = os.path.join(datadir, "external") self.events_oldest_timestamp = utils.current_tai() - def basic_make_csc(self, initial_state, config_dir=None, simulation_mode=0): + def basic_make_csc( + self, initial_state: State, config_dir: str | None = None, simulation_mode: int = 0 + ) -> scriptqueue.ScriptQueue: csc = scriptqueue.ScriptQueue( index=SalIndex.MAIN_TEL, initial_state=initial_state, @@ -247,13 +253,13 @@ def basic_make_csc(self, initial_state, config_dir=None, simulation_mode=0): ) return csc - async def asyncTearDown(self): + async def asyncTearDown(self) -> None: try: await super().asyncTearDown() except AssertionError: pass - def make_stop_data(self, stop_indices, terminate): + def make_stop_data(self, stop_indices: list[int], terminate: bool) -> BaseMsgType: """Make data for the stopScripts command. Parameters @@ -273,13 +279,13 @@ def make_stop_data(self, stop_indices, terminate): async def assert_next_queue( self, - enabled=True, - running=False, - current_sal_index=0, - sal_indices=(), - past_sal_indices=(), - verbose=False, - ): + enabled: bool = True, + running: bool = False, + current_sal_index: int = 0, + sal_indices: Iterable[int] = [], + past_sal_indices: Iterable[int] = [], + verbose: bool = False, + ) -> BaseMsgType: """Get the next queue event and check values. The defaults are appropriate to an enabled, paused queue @@ -368,7 +374,26 @@ async def assert_next_queue( assert list(queue_data.pastSalIndices[0 : queue_data.pastLength]) == list(past_sal_indices) return queue_data - async def assert_next_next_visit(self, sal_index): + async def wait_script_state(self, sal_index: int, script_state: ScriptState) -> BaseMsgType: + """Wait until script process state matches the provided process state. + + Parameters + ---------- + script_state : `ScriptProcessState` + Expected process state. + + Returns + ------- + script : `BaseMsgType` + The sample scriopt event. + """ + data = await self.assert_next_sample(self.remote.evt_script) + while not (data.scriptSalIndex == sal_index and data.scriptState == script_state): + data = await self.assert_next_sample(self.remote.evt_script) + + return data + + async def assert_next_next_visit(self, sal_index: int) -> BaseMsgType: """Assert that the next nextVisit event is for the specified index and return the event data. @@ -386,7 +411,7 @@ async def assert_next_next_visit(self, sal_index): assert data.groupId != "" return data - async def assert_next_next_visit_canceled(self, sal_index): + async def assert_next_next_visit_canceled(self, sal_index: int) -> None: """Assert that the next nextVisitCanceled event is for the specified index. @@ -403,7 +428,7 @@ async def assert_next_next_visit_canceled(self, sal_index): @patch("lsst.ts.utils.ImageNameServiceClient.get_next_obs_id") @patch.dict(os.environ, {"IMAGE_SERVER_URL": "mytemp"}) - async def test_add_block(self, mock_get): + async def test_add_block(self, mock_get: Mock) -> None: """Test adding scripts that are part of a block.""" mock_get.side_effect = [ @@ -411,7 +436,7 @@ async def test_add_block(self, mock_get): (0, ["BL1_O_20240228_000002"]), ] is_standard = False - path = "script1" + path = "script7" config = "wait_time: 1" # give showScript time to run make_add_kwargs = MakeAddKwargs( isStandard=is_standard, path=path, config=config, descr="test_add_block" @@ -501,10 +526,10 @@ async def test_add_block(self, mock_get): assert "BL1_O_20240228_000001" in block_ids assert "BL1_O_20240228_000002" in block_ids - async def test_add_remove(self): + async def test_add_remove(self) -> None: """Test add, remove and showScript.""" is_standard = False - path = "script1" + path = "script7" config = "wait_time: 1" # give showScript time to run make_add_kwargs = MakeAddKwargs( isStandard=is_standard, path=path, config=config, descr="test_add_remove" @@ -698,7 +723,7 @@ async def test_add_remove(self): with pytest.raises(salobj.AckError): await self.remote.cmd_showScript.set_start(scriptSalIndex=3579, timeout=STD_TIMEOUT) - async def check_add_log_level(self, log_level): + async def check_add_log_level(self, log_level: int) -> None: """Test script log level when adding a script to the script queue.""" async with ( self.make_csc(initial_state=salobj.State.ENABLED), @@ -714,7 +739,7 @@ async def check_add_log_level(self, log_level): await self.remote.cmd_add.set_start( logLevel=log_level, isStandard=False, - path="script1", + path="script7", config="", location=Location.LAST, descr="test_add_log_level", @@ -750,7 +775,7 @@ async def check_add_log_level(self, log_level): await self.assert_next_queue(enabled=True, running=True, current_sal_index=I0) await self.assert_next_queue(enabled=True, running=True, past_sal_indices=[I0]) - async def get_next_sample(self, topic): + async def get_next_sample(self, topic: salobj.topics.ReadTopic) -> BaseMsgType: sample = await topic.next(flush=False, timeout=STD_TIMEOUT) while sample.private_sndStamp <= self.events_oldest_timestamp: print(f"Discarding old {sample=}.") @@ -758,14 +783,14 @@ async def get_next_sample(self, topic): return sample - async def check_bin_script_initial_state(self, cmdline_args): + async def check_bin_script_initial_state(self, cmdline_args: Sequence[str]) -> None: for initial_state, index in ( (None, SalIndex.MAIN_TEL), (salobj.State.STANDBY, SalIndex.AUX_TEL), (salobj.State.DISABLED, SalIndex.MAIN_TEL), (salobj.State.ENABLED, SalIndex.AUX_TEL), ): - salobj.set_random_lsst_dds_partition_prefix() + salobj.set_test_topic_subname() with self.subTest(initial_state=initial_state, index=index): await self.check_bin_script( name="ScriptQueue", @@ -775,18 +800,18 @@ async def check_bin_script_initial_state(self, cmdline_args): cmdline_args=cmdline_args, ) - async def test_add_nonzero_log_level(self): + async def test_add_nonzero_log_level(self) -> None: """Test addding a script with a non-zero log level.""" # pick a level that does not match the default # to make it easier to see that the level has changed log_level = logging.INFO - 1 await self.check_add_log_level(log_level=log_level) - async def test_add_zero_log_level(self): + async def test_add_zero_log_level(self) -> None: """Test addding a script with log level 0, meaning don't change it.""" await self.check_add_log_level(log_level=0) - async def test_add_and_pause(self): + async def test_add_and_pause(self) -> None: """Test adding a script with a pause checkpoint.""" async with ( self.make_csc(initial_state=salobj.State.DISABLED), @@ -800,7 +825,7 @@ async def test_add_and_pause(self): await self.remote.cmd_add.set_start( pauseCheckpoint="start", isStandard=False, - path="script1", + path="script7", config="", location=Location.LAST, descr="test_add", @@ -828,7 +853,7 @@ async def test_add_and_pause(self): await script_remote.cmd_resume.start(timeout=STD_TIMEOUT) await self.assert_next_queue(running=True, past_sal_indices=[I0]) - async def test_add_and_stop(self): + async def test_add_and_stop(self) -> None: """Test adding a script with a stop checkpoint.""" async with ( self.make_csc(initial_state=salobj.State.DISABLED), @@ -842,7 +867,7 @@ async def test_add_and_stop(self): await self.remote.cmd_add.set_start( stopCheckpoint="start", isStandard=False, - path="script1", + path="script7", config="", location=Location.LAST, descr="test_add", @@ -867,7 +892,7 @@ async def test_add_and_stop(self): await self.assert_next_queue(running=True, past_sal_indices=[I0]) - async def test_bin_script_state_with_test_scripts(self): + async def test_bin_script_state_with_test_scripts(self) -> None: """Test the --state argument of run_script_queue Note that other bin script tests are in a separate class below, @@ -887,7 +912,7 @@ async def test_bin_script_state_with_test_scripts(self): standardscripts is None or externalscripts is None, "Could not import ts_standardscripts and/or ts_externalscripts.", ) - async def test_bin_script_state(self): + async def test_bin_script_state(self) -> None: """Test the --state argument of run_script_queue Note that other bin script tests are in a separate class below, @@ -898,7 +923,7 @@ async def test_bin_script_state(self): """ await self.check_bin_script_initial_state(cmdline_args=()) - async def test_process_state(self): + async def test_script_state(self) -> None: """Test the processState value of the queue event.""" async with self.make_csc(initial_state=salobj.State.DISABLED): await self.assert_next_sample( @@ -907,7 +932,7 @@ async def test_process_state(self): subsystemVersions="", ) - make_add_kwargs = MakeAddKwargs(descr="test_process_state") + make_add_kwargs = MakeAddKwargs(descr="test_script_state") await self.assert_next_queue(enabled=False, running=True) @@ -1015,7 +1040,7 @@ async def test_process_state(self): assert script_data2.processState == ScriptProcessState.DONE assert script_data2.scriptState == ScriptState.DONE - async def test_unloadable_script(self): + async def test_unloadable_script(self) -> None: """Test adding a script that fails while loading.""" async with self.make_csc(initial_state=salobj.State.DISABLED): await self.assert_next_queue(enabled=False, running=True) @@ -1043,7 +1068,7 @@ async def test_unloadable_script(self): await self.assert_next_queue(enabled=True, running=True, past_sal_indices=[I0]) - async def test_move(self): + async def test_move(self) -> None: """Test move, pause and showQueue""" async with self.make_csc(initial_state=salobj.State.DISABLED): await self.assert_next_queue(enabled=False, running=True) @@ -1206,7 +1231,7 @@ async def test_move(self): queue_data = await self.remote.evt_queue.next(flush=False, timeout=STD_TIMEOUT) assert queue_data.length == 0 - async def test_requeue(self): + async def test_requeue(self) -> None: """Test requeue, move and terminate""" async with self.make_csc(initial_state=salobj.State.DISABLED): await self.assert_next_queue(enabled=False, running=True) @@ -1390,7 +1415,7 @@ async def test_requeue(self): past_sal_indices=[I0 + 9, I0 + 2, I0 + 1] + stopped_scripts, ) - async def test_next_visit_canceled(self): + async def test_next_visit_canceled(self) -> None: """Test the nextVisitCanceled event.""" async with self.make_csc(initial_state=salobj.State.DISABLED): make_add_kwargs = MakeAddKwargs(descr="test_next_visit_canceled") @@ -1460,7 +1485,180 @@ async def test_next_visit_canceled(self): past_sal_indices={I0, I0 + 2, I0 + 3, I0 + 1}, ) - async def test_show_available_scripts(self): + async def test_next_visit_start_time(self) -> None: + """Test the next visit event start time feature.""" + async with self.make_csc(initial_state=salobj.State.DISABLED): + make_add_kwargs = MakeAddKwargs(descr="test_next_visit_start_time") + await self.assert_next_queue(enabled=False, running=True) + + await self.remote.cmd_enable.start(timeout=STD_TIMEOUT) + await self.assert_next_queue(enabled=True, running=True) + + # Pause the queue so we know what to expect of queue state. + await self.remote.cmd_pause.start(timeout=STD_TIMEOUT) + await self.assert_next_queue(running=False) + + # Queue 4 scripts: enough to test different start times. + sal_indices = [] + wait_times = [10, 15, 21, 25] + for i in range(4): + sal_indices.append(I0 + i) + add_kwargs = make_add_kwargs(config=f"wait_time: {wait_times[i]}") + await self.remote.cmd_add.set_start(**add_kwargs, timeout=STD_TIMEOUT) + await self.assert_next_queue(sal_indices=sal_indices) + + await self.wait_configured(*sal_indices) + await self.remote.cmd_resume.start(timeout=STD_TIMEOUT) + await self.assert_next_queue(running=True, sal_indices=sal_indices) + + next_visit = await self.assert_next_next_visit(sal_index=I0) + await self.assert_next_queue( + running=True, + current_sal_index=I0, + sal_indices=[I0 + 1, I0 + 2, I0 + 3], + ) + script_data = await self.wait_script_state(sal_index=I0, script_state=ScriptState.RUNNING) + assert next_visit.startTime == pytest.approx(script_data.timestampRunStart, abs=1.0) + + next_visit = await self.assert_next_next_visit(sal_index=I0 + 1) + await self.assert_next_queue( + running=True, + current_sal_index=I0 + 1, + sal_indices=[I0 + 2, I0 + 3], + past_sal_indices=[I0], + ) + script_data = await self.wait_script_state(sal_index=I0 + 1, script_state=ScriptState.RUNNING) + assert next_visit.startTime == pytest.approx(script_data.timestampRunStart, abs=1.0) + + next_visit = await self.assert_next_next_visit(sal_index=I0 + 2) + await self.assert_next_queue( + running=True, + current_sal_index=I0 + 2, + sal_indices=[I0 + 3], + past_sal_indices=[I0 + 1, I0], + ) + script_data = await self.wait_script_state(sal_index=I0 + 2, script_state=ScriptState.RUNNING) + assert next_visit.startTime == pytest.approx(script_data.timestampRunStart, abs=1.0) + + next_visit = await self.assert_next_next_visit(sal_index=I0 + 3) + await self.assert_next_queue( + running=True, + current_sal_index=I0 + 3, + sal_indices=[], + past_sal_indices=[I0 + 2, I0 + 1, I0], + ) + script_data = await self.wait_script_state(sal_index=I0 + 3, script_state=ScriptState.RUNNING) + assert next_visit.startTime == pytest.approx(script_data.timestampRunStart, abs=1.0) + + await self.assert_next_queue( + running=True, + current_sal_index=0, + sal_indices=[], + past_sal_indices=[I0 + 3, I0 + 2, I0 + 1, I0], + ) + + # wait a few seconds and add another script. + await asyncio.sleep(5) + sal_indices.append(I0 + 4) + add_kwargs = make_add_kwargs(config="wait_time: 10") + await self.remote.cmd_add.set_start(**add_kwargs, timeout=STD_TIMEOUT) + + next_visit = await self.assert_next_next_visit(sal_index=I0 + 4) + await self.assert_next_queue( + running=True, + current_sal_index=0, + sal_indices=[I0 + 4], + past_sal_indices=[I0 + 3, I0 + 2, I0 + 1, I0], + ) + await self.assert_next_queue( + running=True, + current_sal_index=I0 + 4, + sal_indices=[], + past_sal_indices=[I0 + 3, I0 + 2, I0 + 1, I0], + ) + script_data = await self.wait_script_state(sal_index=I0 + 4, script_state=ScriptState.RUNNING) + assert next_visit.startTime == pytest.approx(script_data.timestampRunStart, abs=1.0) + + await self.assert_next_queue( + running=True, + current_sal_index=0, + sal_indices=[], + past_sal_indices=[I0 + 4, I0 + 3, I0 + 2, I0 + 1, I0], + ) + + # I will now add 3 scripts, wait until the first one executes for a + # bit, then stop the script at the top of the waiting queue, the + # next visit. + + past_sal_indices = list(range(I0 + 4, I0 - 1, -1)) + + # Pause the queue so we know what to expect of queue state. + await self.remote.cmd_pause.start(timeout=STD_TIMEOUT) + await self.assert_next_queue( + running=False, + past_sal_indices=past_sal_indices, + ) + + # Queue 3 scripts + sal_indices = [] + wait_times = [10, 15, 20] + for i in range(3): + sal_indices.append(I0 + 5 + i) + add_kwargs = make_add_kwargs(config=f"wait_time: {wait_times[i]}") + await self.remote.cmd_add.set_start(**add_kwargs, timeout=STD_TIMEOUT) + await self.assert_next_queue( + sal_indices=sal_indices, + past_sal_indices=past_sal_indices, + ) + + await self.wait_configured(*sal_indices) + await self.remote.cmd_resume.start(timeout=STD_TIMEOUT) + next_visit = await self.assert_next_next_visit(sal_index=I0 + 5) + await self.assert_next_queue( + running=True, + sal_indices=sal_indices, + past_sal_indices=past_sal_indices, + ) + + await self.assert_next_next_visit(sal_index=I0 + 6) + await self.assert_next_queue( + running=True, + current_sal_index=I0 + 5, + sal_indices=[I0 + 6, I0 + 7], + past_sal_indices=past_sal_indices, + ) + script_data = await self.wait_script_state(sal_index=I0 + 5, script_state=ScriptState.RUNNING) + assert next_visit.startTime == pytest.approx(script_data.timestampRunStart, abs=1.0) + + stop_data = self.make_stop_data(stop_indices=[I0 + 6], terminate=False) + await self.remote.cmd_stopScripts.start(stop_data, timeout=STD_TIMEOUT) + await self.assert_next_next_visit_canceled(sal_index=I0 + 6) + next_visit = await self.assert_next_next_visit(sal_index=I0 + 7) + past_sal_indices.insert(0, I0 + 6) + await self.assert_next_queue( + running=True, + current_sal_index=I0 + 5, + sal_indices=[I0 + 7], + past_sal_indices=past_sal_indices, + ) + + past_sal_indices.insert(0, I0 + 5) + await self.assert_next_queue( + running=True, + current_sal_index=I0 + 7, + sal_indices=[], + past_sal_indices=past_sal_indices, + ) + script_data = await self.wait_script_state(sal_index=I0 + 7, script_state=ScriptState.RUNNING) + assert next_visit.startTime == pytest.approx(script_data.timestampRunStart, abs=1.0) + + past_sal_indices.insert(0, I0 + 7) + await self.assert_next_queue( + running=True, + past_sal_indices=past_sal_indices, + ) + + async def test_show_available_scripts(self) -> None: """Test the showAvailableScripts command.""" async with self.make_csc(initial_state=salobj.State.DISABLED): # Make sure showAvailableScripts fails when not enabled. @@ -1489,7 +1687,7 @@ async def test_show_available_scripts(self): "subdir/subsubdir/script4", ] ) - expected_ext_set = set(["script1", "script5", "subdir/script3", "subdir/script6"]) + expected_ext_set = set(["script7", "script5", "subdir/script8", "subdir/script6"]) for available_scripts in (available_scripts0, available_scripts1): standard_set = set(available_scripts.standard.split(":")) external_set = set(available_scripts.external.split(":")) @@ -1502,11 +1700,11 @@ async def test_show_available_scripts(self): with pytest.raises(salobj.AckError): await self.remote.cmd_showAvailableScripts.start(timeout=STD_TIMEOUT) - async def test_show_schema(self): + async def test_show_schema(self) -> None: """Test the showSchema command.""" async with self.make_csc(initial_state=salobj.State.DISABLED): is_standard = False - path = "script1" + path = "script7" await self.assert_next_queue(enabled=False, running=True) self.remote.cmd_showSchema.set(isStandard=is_standard, path=path) @@ -1524,7 +1722,7 @@ async def test_show_schema(self): schema = yaml.safe_load(data.configSchema) assert schema == salobj.TestScript.get_schema() - async def test_show_queue(self): + async def test_show_queue(self) -> None: """Test the showQueue command.""" async with self.make_csc(initial_state=salobj.State.DISABLED): await self.assert_next_queue(enabled=False, running=True) @@ -1552,7 +1750,7 @@ async def test_show_queue(self): with pytest.raises(salobj.AckError): await self.remote.cmd_showQueue.start(timeout=STD_TIMEOUT) - async def wait_configured(self, *sal_indices): + async def wait_configured(self, *sal_indices: int) -> None: """Wait for the specified scripts to be configured. Call this before running the queue if you want the queue data @@ -1569,8 +1767,8 @@ async def wait_configured(self, *sal_indices): class CmdLineTestCase(unittest.IsolatedAsyncioTestCase): - def setUp(self): - salobj.set_random_lsst_dds_partition_prefix() + def setUp(self) -> None: + salobj.set_test_topic_subname() self.index = 1 try: self.default_standardpath = scriptqueue.get_default_scripts_dir(is_standard=True) @@ -1588,7 +1786,7 @@ def setUp(self): self.badpath = os.path.join(self.datadir, "not_a_directory") self.events_oldest_timestamp = utils.current_tai() - async def get_next_sample(self, topic): + async def get_next_sample(self, topic: salobj.topics.ReadTopic) -> BaseMsgType: sample = await topic.next(flush=False, timeout=STD_TIMEOUT) while sample.private_sndStamp <= self.events_oldest_timestamp: print(f"Discarding old {sample=}.") @@ -1596,7 +1794,7 @@ async def get_next_sample(self, topic): return sample - async def test_run_with_standard_and_external(self): + async def test_run_with_standard_and_external(self) -> None: exe_name = "run_script_queue" exe_path = shutil.which(exe_name) if exe_path is None: @@ -1640,7 +1838,7 @@ async def test_run_with_standard_and_external(self): standardscripts is None or externalscripts is None, "Could not import ts_standardscripts and/or ts_externalscripts.", ) - async def test_run_default_standard_external(self): + async def test_run_default_standard_external(self) -> None: exe_name = "run_script_queue" exe_path = shutil.which(exe_name) if exe_path is None: diff --git a/tests/test_utils.py b/tests/test_utils.py index 3b2c58f..d66a6aa 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -36,7 +36,7 @@ class UtilsTestCase(unittest.TestCase): - def test_find_public_scripts(self): + def test_find_public_scripts(self) -> None: root = os.path.join(os.path.dirname(__file__), "data/standard") scripts = scriptqueue.find_public_scripts(root) expectedscripts = set( @@ -51,14 +51,14 @@ def test_find_public_scripts(self): assert set(scripts) == expectedscripts @unittest.skipIf(standardscripts is None, "Could not import ts_standardscripts") - def test_get_default_standard_scripts_dir(self): + def test_get_default_standard_scripts_dir(self) -> None: standard_dir = scriptqueue.get_default_scripts_dir(is_standard=True) assert isinstance(standard_dir, pathlib.Path) assert standard_dir.samefile(standardscripts.get_scripts_dir()) assert standard_dir.name == "scripts" @unittest.skipIf(externalscripts is None, "Could not import ts_externalscripts") - def test_get_default_external_scripts_dir(self): + def test_get_default_external_scripts_dir(self) -> None: external_dir = scriptqueue.get_default_scripts_dir(is_standard=False) assert isinstance(external_dir, pathlib.Path) assert external_dir.samefile(externalscripts.get_scripts_dir())