"""Custom command classes for commands requiring overrides to do()."""
import functools
import logging
from typing import Any, Optional
from ska_control_model import ResultCode, TaskStatus
from ska_tango_base.base import CommandTracker
from ska_tango_base.commands import FastCommand, SlowCommand, SubmittedSlowCommand
from ska_mid_dish_manager.component_managers.dish_manager_cm import DishManagerComponentManager
[docs]class AbortCommand(SlowCommand):
"""A custom class for Abort Command."""
[docs] def __init__(
self, command_tracker, component_manager, callback, logger: logging.Logger | None = None
) -> None:
"""Initialise a new AbortCommand instance.
:param command_tracker: the device's command tracker
:param component_manager: the device's component manager
:param callback: callback to be called when this command
starts and finishes
:param logger: a logger for this command object to use
"""
self._command_tracker: CommandTracker = command_tracker
self._component_manager: DishManagerComponentManager = component_manager
# _callback is an instance of Abort in abort_sequence_command_handler.py
super().__init__(callback=callback, logger=logger)
[docs] def do(self, *args: Any, **kwargs: Any) -> tuple[ResultCode, str]:
"""Stateless hook for command functionality.
:param args: positional args to the component manager method
:param kwargs: keyword args to the component manager method
:return: A tuple containing the result code (e.g. STARTED)
and a string message containing a command_id (if
the command has been accepted) or an informational message
(if the command was rejected)
"""
abort_sequence_command_id = self._command_tracker.new_command("Abort")
# replace the command_id in the callback with the abort_sequence_command_id
# so that clients tracking the long running command result receive updates
# only when the abort sequence command completes.
self._callback.command_id = abort_sequence_command_id
abort_task_command_id = self._command_tracker.new_command(
"cancel-lrcs", completed_callback=self._completed
)
status, message = self._component_manager.abort(
functools.partial(self._command_tracker.update_command_info, abort_task_command_id),
)
if status == TaskStatus.IN_PROGRESS:
return ResultCode.STARTED, abort_sequence_command_id
if status == TaskStatus.REJECTED:
return ResultCode.REJECTED, abort_sequence_command_id
return (
ResultCode.FAILED,
f"{status.name} was returned by command method with message: {message}",
)
[docs]class ApplyPointingModelCommand(FastCommand):
"""Class for handling band pointing parameters given a JSON input."""
[docs] def __init__(self, component_manager, logger: Optional[logging.Logger] = None) -> None:
"""Initialise a new ApplyPointingModelCommand instance.
:param component_manager: the device to which this command belongs.
:param logger: a logger for this command to use.
"""
self._component_manager = component_manager
super().__init__(logger)
[docs] def do(self, *args: Any, **kwargs: Any) -> tuple[ResultCode, str]:
"""Implement ApplyPointingModel command functionality.
:param args: JSON object with a schema similar to this::
{
"interface": "...",
"antenna": "....",
"band": "`Band_`...",
"attrs": {...},
"coefficients": {
"IA": {...},
...
...
"HESE8":{...}
},
"rms_fits":
{
"xel_rms": {...},
"el_rms": {...},
"sky_rms": {...}
}
}
:return: A tuple containing a return code and a string
message indicating status.
"""
return self._component_manager.apply_pointing_model(*args)
[docs]class SetKValueCommand(FastCommand):
"""Class for handling the SetKValue command."""
[docs] def __init__(self, component_manager, logger: Optional[logging.Logger] = None) -> None:
"""Initialise a new SetKValueCommand instance.
:param component_manager: the device to which this command belongs.
:param logger: a logger for this command to use.
"""
self._component_manager = component_manager
super().__init__(logger)
[docs] def do(self, *args: Any, **kwargs: Any) -> tuple[ResultCode, str]:
"""Implement SetKValue command functionality.
:param args: k value.
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
return self._component_manager.set_kvalue(*args)
[docs]class ResetComponentConnectionCommand(FastCommand):
"""Class for handling the ResetComponentConnection command."""
[docs] def __init__(self, component_manager, logger: Optional[logging.Logger] = None) -> None:
"""Initialise a new ResetComponentConnectionCommand instance.
:param component_manager: the device to which this command belongs.
:param logger: a logger for this command to use.
"""
self._component_manager = component_manager
super().__init__(logger)
[docs] def do(self, *args: Any, **kwargs: Any) -> tuple[ResultCode, str]:
"""Implement ResetComponentConnection command functionality.
:param args: list of Subservient device names.
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
return self._component_manager.reset_subservient_dev_connections(*args)
[docs]class SetFrequencyCommand(FastCommand):
"""Class for handling the SetFrequency command."""
[docs] def __init__(self, component_manager, logger: Optional[logging.Logger] = None) -> None:
"""Initialise a new SetFrequency instance.
:param component_manager: the device to which this command belongs.
:param logger: a logger for this command to use.
"""
self._component_manager = component_manager
super().__init__(logger)
[docs] def do(self, *args: Any, **kwargs: Any) -> tuple[ResultCode, str]:
"""Implement SetFrequency command functionality.
:param args: PLL output frequency (GHz)
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
return self._component_manager.set_frequency(*args)
[docs]class SetVPolAttenuationCommand(FastCommand):
"""Class for handling the SetVPolAttenuation command."""
[docs] def __init__(self, component_manager, logger: Optional[logging.Logger] = None) -> None:
"""Initialise a new SetVPolAttenuation instance.
:param component_manager: the device to which this command belongs.
:param logger: a logger for this command to use.
"""
self._component_manager = component_manager
super().__init__(logger)
[docs] def do(self, *args: Any, **kwargs: Any) -> tuple[ResultCode, str]:
"""Implement SetVPolAttenuation command functionality.
:param args: RFCM V-polarization attenuation value (dB)
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
return self._component_manager.set_v_pol_attenuation(*args)
[docs]class SetHPolAttenuationCommand(FastCommand):
"""Class for handling the SetHPolAttenuation command."""
[docs] def __init__(self, component_manager, logger: Optional[logging.Logger] = None) -> None:
"""Initialise a new SetHPolAttenuation instance.
:param component_manager: the device to which this command belongs.
:param logger: a logger for this command to use.
"""
self._component_manager = component_manager
super().__init__(logger)
[docs] def do(self, *args: Any, **kwargs: Any) -> tuple[ResultCode, str]:
"""Implement SetHPolAttenuation command functionality.
:param args: RFCM H-polarization attenuation value (dB)
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
return self._component_manager.set_h_pol_attenuation(*args)
[docs]class ResetTrackTableCommand(FastCommand):
"""Class for handling the ResetTrackTable command."""
[docs] def __init__(self, component_manager, logger: Optional[logging.Logger] = None) -> None:
"""Initialise a new ResetTrackTableCommand instance.
:param component_manager: the device to which this command belongs.
:param logger: a logger for this command to use.
"""
self._component_manager = component_manager
super().__init__(logger)
[docs] def do(self, *args: Any, **kwargs: Any) -> tuple[ResultCode, str]:
"""Implement ResetTrackTable command functionality.
:return: A tuple containing a return code and a string
message indicating status.
"""
return self._component_manager.reset_track_table(*args)
[docs]class StowCommand(SubmittedSlowCommand):
"""A custom class for Stow Command."""
[docs] def do(self: SubmittedSlowCommand, *args: Any, **kwargs: Any) -> tuple[ResultCode, str]:
"""Stateless hook for command functionality.
:param args: positional args to the component manager method
:param kwargs: keyword args to the component manager method
:return: A tuple containing the result code (e.g. STARTED)
and a string message if the command has been accepted or failed
"""
command_id = self._command_tracker.new_command(
self._command_name, completed_callback=self._completed
)
method = getattr(self._component_manager, self._method_name)
status, message = method(
*args,
functools.partial(self._command_tracker.update_command_info, command_id),
**kwargs,
)
# the stow command is not tracked from this point
# dont return the command id to the client
if status == TaskStatus.COMPLETED:
return ResultCode.STARTED, "Stow called on Dish Structure, monitor dishmode for STOW"
elif status == TaskStatus.IN_PROGRESS:
return (
ResultCode.STARTED,
"Stow will be called on Dish Structure, monitor dishmode for STOW",
)
return (
ResultCode.FAILED,
f"{status.name} was returned by command method with message: {message}",
)