"""A submitted slow command with do() overridden."""
import functools
import logging
from typing import Any, Optional
from ska_control_model import ResultCode, TaskStatus
from ska_tango_base.commands import FastCommand, SubmittedSlowCommand
from ska_mid_dish_ds_manager.opcua_component_manager import OPCUAComponentManager
[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 stow method with message: {message}",
)
[docs]class ReleaseAuthCommand(SubmittedSlowCommand):
"""A custom class for ReleaseAuth 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. OK)
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 release auth command is not tracked from this point
# dont return the command id to the client
if status == TaskStatus.COMPLETED:
return ResultCode.OK, "ReleaseAuth requested on Dish Structure Controller"
return (
ResultCode.FAILED,
f"{status.name} was returned by release_auth method with message: {message}",
)
[docs]class TrackLoadTableCommand(FastCommand):
"""Class for handling the TrackLoadTable command."""
[docs] def __init__(
self,
component_manager: OPCUAComponentManager,
logger: Optional[logging.Logger] = None,
) -> None:
"""Initialise a new TrackLoadTableCommand 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 TrackLoadTable command functionality.
:param args: table.
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
return self._component_manager.command_manager.track_load_table(*args)