Developer guide to MccsCommandProxy.

This guide is aimed towards developers of MCCS looking to implement the MccsCommandProxy or MccsCompositeCommandProxy.

MccsCommandProxy

The MccsCommandProxy is used to invoke a command on a device using a throwaway proxy

The command proxy interacts with the device interface, hiding its details from the user, and calls the task_callback as appropriate. For example the below snippet will construct a proxy to the device using the device_name and exectute a LongRunningCommand (LRC) waiting for the command to finish.

solve_command = MccsCommandProxy(
    device_name=self._station_calibration_solver_name,
    command_name="Solve",
    logger=self.logger,
)
result, message = solve_command(
    is_lrc=True, arg=json.dumps(kwargs), wait_for_result=True
)

Warning

Warning

When using asynchronous mode (i.e., is_lrc=True and wait_for_result=False), you must retain a reference to the MccsCommandProxy instance for as long as the command is running.

If the object is garbage-collected, the internal event subscription may be lost, and the callback will not be triggered as expected.

MccsCompositeCommandProxy

The MccsCompositeCommandProxy is a collection of MccsCommandProxys. This is a blocking function and will wrap up the results of a MccsCommandProxy using a number of CommandRules to evaluate against. If you choose to supply no CommandRules the MccsCompositeCommandProxy will return ResultCode.FAILED unless all the MccsCommandProxys return ResultCode.OK.

For example, if you are looking at implementing MccsStation.On(). As a developer you know this will call FieldStation.On() and SpsStation.On(). You may want to allow the possible return (ResultCode.REJECTED, “Device is already in ON state.”) as an allowed reply. The CommandRule.allowed_result is an optional list containing a tuple[tuple[ResultCode, str]]. Sometimes the exact string will be dynamic, as a result we will check that this string is a subset of the command result.

def _on(
    self: StationComponentManager,
    task_callback: Optional[Callable] = None,
    task_abort_event: Optional[threading.Event] = None,
) -> None:
    """
    Turn on this station.

    The order to turn a station on is: FieldStation, then tiles and
    antennas.

    :param task_callback: Update task state, defaults to None
    :param task_abort_event: Abort the task
    """
    composite_on = MccsCompositeCommandProxy(self.logger)
    rules = [
        CommandRule(
            self._field_station_trl,
            "On",
            [(ResultCode.REJECTED, "Device is already in ON state.")],
        )
    ]
    composite_on += MccsCommandProxy(self._field_station_trl, "On", self.logger)
    composite_on += MccsCommandProxy(self._sps_station_trl, "On", self.logger)
    result_evaluator = CompositeCommandResultEvaluator(rules)
    composite_on(task_callback=task_callback, command_evaluator=result_evaluator, timeout=600)