# -*- coding: utf-8 -*-
#
# This file is part of the SKA Low MCCS project
#
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE for more info.
"""This module implements the MCCS subarray beam device."""
from __future__ import annotations # allow forward references in type hints
import functools
import importlib
import json
import logging
import sys
from typing import Any, Callable, Final, Optional
from ska_control_model import (
CommunicationStatus,
HealthState,
ObsState,
ObsStateModel,
PowerState,
ResultCode,
TaskStatus,
)
from ska_tango_base.base import CommandTracker
from ska_tango_base.commands import JsonValidator, SlowCommand, SubmittedSlowCommand
from ska_tango_base.obs import SKAObsDevice
from tango.server import attribute, command, device_property
from ska_low_mccs.subarray_beam.subarray_beam_component_manager import (
SubarrayBeamComponentManager,
)
from ska_low_mccs.subarray_beam.subarray_beam_health_model import (
SubarrayBeamHealthModel,
)
__all__ = ["MccsSubarrayBeam", "main"]
DevVarLongStringArrayType = tuple[list[ResultCode], list[Optional[str]]]
# pylint: disable=too-many-public-methods
[docs]class MccsSubarrayBeam(SKAObsDevice):
"""An implementation of a subarray beam Tango device for MCCS."""
ObsCommandTimeout = device_property(
dtype=int,
default_value=60,
doc="The timeout in seconds for Observation commands.",
)
# ---------------
# Initialisation
# ---------------
[docs] def __init__(self, *args: Any, **kwargs: Any) -> None:
"""
Initialise this device object.
:param args: positional args to the init
:param kwargs: keyword args to the init
"""
# We aren't supposed to define initialisation methods for Tango
# devices; we are only supposed to define an `init_device` method. But
# we insist on doing so here, just so that we can define some
# attributes, thereby stopping the linters from complaining about
# "attribute-defined-outside-init" etc. We still need to make sure that
# `init_device` re-initialises any values defined in here.
super().__init__(*args, **kwargs)
self.component_manager: SubarrayBeamComponentManager
self._health_state: HealthState = HealthState.OK
self._health_model: SubarrayBeamHealthModel
self.obs_state_model: ObsStateModel
self._missed_events: int
[docs] def init_device(self: MccsSubarrayBeam) -> None:
"""Initialise the device."""
self._missed_events = 0
super().init_device()
self._build_state = sys.modules["ska_low_mccs"].__version_info__
self._version_id = sys.modules["ska_low_mccs"].__version__
device_name = f'{str(self.__class__).rsplit(".", maxsplit=1)[-1][0:-2]}'
version = f"{device_name} Software Version: {self._version_id}"
self.logger.info("\n%s\n%s", str(self.GetVersionInfo()), version)
def _init_state_model(self: MccsSubarrayBeam) -> None:
super()._init_state_model()
self._health_state = (
HealthState.OK
) # Beam starts EMPTY, which is OK. InitCommand.do() does this too late.
self._health_model = SubarrayBeamHealthModel(
[], self._health_changed, ignore_power_state=True
)
self.set_change_event("healthState", True, False)
self.obs_state_model = ObsStateModel(
logger=self.logger, callback=self._update_obs_state
)
[docs] def create_component_manager(
self: MccsSubarrayBeam,
) -> SubarrayBeamComponentManager:
"""
Create and return a component manager for this device.
:return: a component manager for this device.
"""
return SubarrayBeamComponentManager(
self.logger,
self.ObsCommandTimeout,
self._communication_state_callback,
self._component_state_callback,
)
# commands with a json schema
schema_assign_resources: Final = json.loads(
importlib.resources.read_text(
"ska_low_mccs.schemas.subarray_beam",
"MccsSubarrayBeam_AssignResources_3_0.json",
)
)
schema_configure: Final = json.loads(
importlib.resources.read_text(
"ska_low_mccs.schemas.subarray_beam",
"MccsSubarrayBeam_Configure_3_0.json",
)
)
schema_scan: Final = json.loads(
importlib.resources.read_text(
"ska_low_mccs.schemas.common",
"Mccs_Scan_3_0.json",
)
)
[docs] def init_command_objects(self: MccsSubarrayBeam) -> None:
"""Initialise the command handlers for commands supported by this device."""
super().init_command_objects()
def _callback(hook: str, running: bool) -> None:
action = "invoked" if running else "completed"
full_action = f"{hook}_{action}"
self.logger.debug(full_action)
self.obs_state_model.perform_action(full_action)
# commands with parameters defined by a Json schema specify a validator
# commands which affect the obsState specify a callback on entry/exit
for command_name, method_name, schema, state_model_hook in [
(
"AssignResources",
"assign_resources",
self.schema_assign_resources,
"assign",
),
("ReleaseAllResources", "release_all_resources", None, "release"),
("Configure", "configure", self.schema_configure, "configure"),
("Scan", "scan", self.schema_scan, None),
("EndScan", "end_scan", None, None),
("End", "end", None, None),
# ("Abort", "abort", None, "abort"),
("ObsReset", "obsreset", None, "obsreset"),
("Restart", "restart", None, "restart"),
]:
callback = (
None
if state_model_hook is None
else functools.partial(_callback, state_model_hook)
)
validator = (
None
if schema is None
else JsonValidator(
command_name,
schema,
logger=self.logger,
)
)
self.register_command_object(
command_name,
SubmittedSlowCommand(
command_name,
self._command_tracker,
self.component_manager,
method_name,
callback=callback,
logger=self.logger,
validator=validator,
),
)
self.register_command_object(
"Abort",
self.AbortCommand(
self._command_tracker,
self.component_manager,
callback=functools.partial(_callback, "abort"),
logger=self.logger,
),
)
self.register_command_object(
"AbortDevice",
self.AbortDeviceCommand(
self._command_tracker,
self.component_manager,
callback=functools.partial(_callback, "abort"),
logger=self.logger,
),
)
[docs] class InitCommand(SKAObsDevice.InitCommand):
"""
A class for :py:class:`~.MccsSubarrayBeam`'s Init command.
The :py:meth:`~.MccsSubarrayBeam.InitCommand.do` method below is
called upon :py:class:`~.MccsSubarrayBeam`'s initialisation.
"""
# pylint: disable-next=arguments-differ
[docs] def do( # type: ignore[override]
self: MccsSubarrayBeam.InitCommand,
) -> tuple[ResultCode, str]:
"""
Initialise the attributes and properties of the MccsSubarrayBeam.
State is managed under the hood; the basic sequence is:
1. Device state is set to INIT
2. The do() method is run
3. Device state is set to the OFF
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
self._device._build_state = sys.modules["ska_low_mccs"].__version_info__
self._device._version_id = sys.modules["ska_low_mccs"].__version__
super().do()
return (ResultCode.OK, "Initialisation complete")
# ----------
# Callbacks
# ----------
# pylint: disable=too-many-arguments, too-many-branches
def _component_state_callback(
self: MccsSubarrayBeam,
power: Optional[PowerState] = None,
obsfault: Optional[bool] = None,
health: Optional[HealthState] = None,
beam_locked: Optional[bool] = None,
resources_changed: Optional[set] = None,
configured_changed: Optional[bool] = None,
scanning_changed: Optional[bool] = None,
trl: Optional[str] = None,
obsstate_changed: Optional[ObsState] = None,
missed_event: Optional[bool] = None,
) -> None:
"""
Handle change in the state of the component.
This is a callback hook, called by the component manager when
the state of the component changes.
:param power: An optional parameter with the new power state of
this device.
:param obsfault: An optional flag if the device is entering or
exiting a fault state.
:param health: An optional parameter with the new health state
of this device.
:param beam_locked: An optional flag with the new beam locked
state of this device.
:param resources_changed: An optional parameter updating the resources for
this SubarrayBeam.
:param configured_changed: An optional flag indicating that this SubarrayBeam's
configuration has changed.
:param scanning_changed: An optional flag indicating that this SubarrayBeam's
scanning status has changed.
:param trl: An optional flag indicating the TRL of the the subdevice
:param obsstate_changed: An optional parameter with the new ObsState of
a subservient device.
:param missed_event: whether the component manager has detected a missed change
event.
"""
if trl is None:
if health is not None:
if self._health_state != health:
self._health_state = health
self.push_change_event("healthState", health)
else:
if health is not None:
self._health_model.station_beam_health_changed(trl, health)
if obsstate_changed is not None:
if obsstate_changed == ObsState.FAULT:
self.component_manager._device_obs_state_fault(
trl, obsstate_changed
)
else:
self.component_manager._device_obs_state_changed(
trl, obsstate_changed
)
if beam_locked is not None:
self._health_model.is_beam_locked_changed(beam_locked)
if resources_changed is not None:
station_beam_trls = resources_changed
self._resources_changed(station_beam_trls)
if configured_changed is not None:
self._configuration_changed(configured_changed)
if scanning_changed is not None:
self._scanning_changed(scanning_changed)
if obsfault is not None:
self.obs_state_model.perform_action("component_obsfault")
if missed_event:
self._missed_events += 1
def _communication_state_callback(
self: MccsSubarrayBeam,
communication_state: CommunicationStatus,
) -> None:
"""
Handle change in communications status between component manager and component.
This is a callback hook, called by the component manager when
the communications status changes. It is implemented here to
drive the op_state.
:param communication_state: the status of communications between
the component manager and its component.
"""
action_map = {
CommunicationStatus.DISABLED: "component_disconnected",
CommunicationStatus.NOT_ESTABLISHED: "component_unknown",
CommunicationStatus.ESTABLISHED: "component_on", # always-on device
}
action = action_map[communication_state]
if action is not None:
self.op_state_model.perform_action(action)
self.logger.debug("Now device is communicating")
self._health_model.update_state(
communicating=communication_state == CommunicationStatus.ESTABLISHED
)
def _configuration_changed(self: MccsSubarrayBeam, is_configured: Any) -> None:
"""
Handle the configuration change.
:param is_configured: if the configuration has changed
"""
if is_configured:
self.obs_state_model.perform_action("component_configured")
else:
self.obs_state_model.perform_action("component_unconfigured")
def _resources_changed(
self: MccsSubarrayBeam,
station_beam_trls: set[str],
) -> None:
"""
Handle change in subarray_beam resources.
This is a callback hook, called by the component manager when
the resources of the subarray_beam changes.
:param station_beam_trls: the TRLs of station beams assigned
to this subarray
"""
if station_beam_trls:
self.logger.debug("component_resourced")
self.obs_state_model.perform_action("component_resourced")
else:
self.logger.debug("component_unresourced")
self.obs_state_model.perform_action("component_unresourced")
self._health_model.update_station_beams(station_beam_trls)
def _scanning_changed(self: MccsSubarrayBeam, is_scanning: bool) -> None:
"""
Handle change in station beam scanning status.
This is a callback hook, called by the component manager when
the scanning status of the station beam changes.
:param is_scanning: if the component is performing a scan
"""
if is_scanning:
self.logger.debug("component_scanning")
self.obs_state_model.perform_action("component_scanning")
else:
self.logger.debug("component_not_scanning")
self.obs_state_model.perform_action("component_not_scanning")
def _health_changed(self: MccsSubarrayBeam, health: HealthState) -> None:
"""
Handle change in this device's health state.
This is a callback hook, called whenever the HealthModel's
evaluated health state changes. It is responsible for updating
the tango side of things i.e. making sure the attribute is up to
date, and events are pushed.
:param health: the new health value
"""
if self._health_state != health:
self._health_state = health
self.push_change_event("healthState", health)
# ----------
# Attributes
# ----------
@attribute(
dtype="DevString",
format="%s",
)
def healthModelParams(self: MccsSubarrayBeam) -> str:
"""
Get the health params from the health model.
:return: the health params
"""
return json.dumps(self._health_model.health_params)
[docs] @healthModelParams.write # type: ignore[no-redef]
def healthModelParams(self: MccsSubarrayBeam, argin: str) -> None:
"""
Set the params for health transition rules.
:param argin: JSON-string of dictionary of health states
"""
self._health_model.health_params = json.loads(argin)
self._health_model.update_health()
[docs] @attribute(dtype="DevLong", format="%i")
def subarrayId(self: MccsSubarrayBeam) -> int:
"""
Return the subarray id.
:return: the subarray id
"""
return self.component_manager.subarray_id
[docs] @attribute(dtype="DevLong", format="%i", max_value=47, min_value=0)
def subarrayBeamId(self: MccsSubarrayBeam) -> int:
"""
Return the subarray beam id.
:return: the subarray beam id
"""
return self.component_manager.subarray_beam_id
@attribute(dtype=(str,), max_dim_x=512)
def stationBeamIds(self: MccsSubarrayBeam) -> list[str]:
"""
Return the ids of station beams assigned to this subarray beam.
:return: the station beam ids
"""
return self.component_manager.station_beam_ids
[docs] @stationBeamIds.write # type: ignore[no-redef]
def stationBeamIds(self: MccsSubarrayBeam, station_beam_ids: list[str]) -> None:
"""
Set the station beam ids.
:param station_beam_ids: ids of the station beams for this subarray beam
"""
self.component_manager.station_beam_ids = station_beam_ids
@attribute(dtype=("DevLong",), max_dim_x=512, format="%i")
def stationIds(self: MccsSubarrayBeam) -> list[int]:
"""
Return the station ids.
:return: the station ids
"""
return self.component_manager.station_ids
[docs] @stationIds.write # type: ignore[no-redef]
def stationIds(self: MccsSubarrayBeam, station_ids: list[int]) -> None:
"""
Set the station ids.
:param station_ids: ids of the stations for this beam
"""
self.component_manager.station_ids = station_ids
@attribute(dtype="DevLong", format="%i", max_value=7, min_value=0)
def logicalBeamId(self: MccsSubarrayBeam) -> int:
"""
Return the logical beam id.
:todo: this documentation needs to differentiate logical beam id
from beam id
:return: the logical beam id
"""
return self.component_manager.logical_beam_id
[docs] @logicalBeamId.write # type: ignore[no-redef]
def logicalBeamId(self: MccsSubarrayBeam, logical_beam_id: int) -> None:
"""
Set the logical beam id.
:param logical_beam_id: the logical beam id
"""
self.component_manager.logical_beam_id = logical_beam_id
[docs] @attribute(
dtype="DevDouble",
unit="Hz",
standard_unit="s^-1",
max_value=1e37,
min_value=0,
)
def updateRate(self: MccsSubarrayBeam) -> float:
"""
Return the update rate (in hertz) for this subarray beam.
:return: the update rate for this subarray beam
"""
return self.component_manager.update_rate
@attribute(dtype="DevBoolean")
def isBeamLocked(self: MccsSubarrayBeam) -> bool:
"""
Return a flag indicating whether the beam is locked or not.
:return: whether the beam is locked or not
"""
return self.component_manager.is_beam_locked
[docs] @isBeamLocked.write # type: ignore[no-redef]
def isBeamLocked(self: MccsSubarrayBeam, value: bool) -> None:
"""
Set a flag indicating whether the beam is locked or not.
:param value: whether the beam is locked or not
"""
self.component_manager.is_beam_locked = value
[docs] @attribute(dtype=(("DevLong",),), max_dim_y=384, max_dim_x=4)
def channels(self: MccsSubarrayBeam) -> list[list[int]]:
"""
Return the ids of the channels configured for this beam.
:return: channel ids
"""
return self.component_manager.channels
[docs] @attribute(dtype=("DevFloat",), max_dim_x=384)
def antennaWeights(self: MccsSubarrayBeam) -> list[float]:
"""
Return the antenna weights configured for this beam.
:return: antenna weightd
"""
return self.component_manager.antenna_weights
@attribute(dtype=("DevDouble",), max_dim_x=5)
def desiredPointing(self: MccsSubarrayBeam) -> list[float]:
"""
Return the desired pointing of this beam.
:return: the desired point of this beam, conforming to the Sky
Coordinate Set definition
"""
return self.component_manager.desired_pointing
[docs] @desiredPointing.write # type:ignore[no-redef]
def desiredPointing(self: MccsSubarrayBeam, values: list[float]) -> None:
"""
Set the desired pointing of this beam.
* activation time (s) -- value range 0-10^37
* azimuth position (deg) -- value range 0-360
* azimuth speed (deg/s) -- value range 0-10^37
* elevation position (deg) -- value range 0-90
* elevation rate (deg/s) -- value range 0-10^37
:param values: the desired pointing of this beam, expressed as a
sky coordinate set
"""
self.component_manager.desired_pointing = values
[docs] @attribute(dtype=("DevDouble",), max_dim_x=5)
def phaseCentre(self: MccsSubarrayBeam) -> list[float]:
"""
Return the phase centre.
:return: the phase centre
"""
return self.component_manager.phase_centre
[docs] @attribute(dtype="DevLong")
def firstSubarrayChannel(self: MccsSubarrayBeam) -> int:
"""
Return the first logical channel allocated to the beam.
:return: the first logical channel allocated to the beam.
"""
return self.component_manager.first_channel
[docs] @attribute(dtype="DevLong")
def numberOfChannels(self: MccsSubarrayBeam) -> int:
"""
Return the first logical channel allocated to the beam.
:return: the first logical channel allocated to the beam.
"""
return self.component_manager.number_of_channels
[docs] @attribute(dtype="DevString")
def healthReport(self: MccsSubarrayBeam) -> str:
"""
Get the health report.
:return: the health report.
"""
return self._health_model.health_report
[docs] @attribute(dtype="DevLong")
def missedEvents(self: MccsSubarrayBeam) -> int:
"""
Get the amount of missed change events.
Some commands rely on change events from sub-devices, sometime we miss these
events, this attribute keeps track of how many we know we have missed.
:return: the amount of missed change events
"""
return self._missed_events
# --------
# Commands
# --------
[docs] @command(dtype_in="DevString", dtype_out="DevVarLongStringArray")
def AssignResources(
self: MccsSubarrayBeam, argin: str
) -> DevVarLongStringArrayType:
"""
Assign resources to the subarray_beam with all relevant parameters.
:param argin: Assign Resources parameters encoded in a json string
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("AssignResources")
(result_code, unique_id) = handler(argin)
return ([result_code], [unique_id])
[docs] @command(dtype_in="DevString", dtype_out="DevVarLongStringArray")
def Scan(self: MccsSubarrayBeam, argin: str) -> DevVarLongStringArrayType:
"""
Start a scan on the subarray_beam.
:param argin: Scan parameters encoded in a json string
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("Scan")
(result_code, unique_id) = handler(argin)
return ([result_code], [unique_id])
[docs] @command(dtype_out="DevVarLongStringArray")
def EndScan(self: MccsSubarrayBeam) -> DevVarLongStringArrayType:
"""
End the scan.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("EndScan")
(result_code, message) = handler()
return ([result_code], [message])
[docs] @command(dtype_out="DevVarLongStringArray")
def ReleaseAllResources(self: MccsSubarrayBeam) -> DevVarLongStringArrayType:
"""
Release all resources.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("ReleaseAllResources")
(result_code, unique_id) = handler()
return ([result_code], [unique_id])
[docs] @command(dtype_out="DevVarLongStringArray")
def End(self: MccsSubarrayBeam) -> DevVarLongStringArrayType:
"""
End the scan block (deconfigure).
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("End")
(result_code, message) = handler()
return ([result_code], [message])
[docs] @command( # type: ignore[misc] # "Untyped decorator makes function untyped"
dtype_out="DevVarLongStringArray",
)
def ObsReset(self: MccsSubarrayBeam) -> DevVarLongStringArrayType:
"""
Reset the current observation process.
To modify behaviour for this command, modify the do() method of
the command class.
:return: A tuple containing a result code and the unique ID of the command
"""
handler = self.get_command_object("ObsReset")
(result_code, message) = handler()
return ([result_code], [message])
[docs] @command( # type: ignore[misc] # "Untyped decorator makes function untyped"
dtype_out="DevVarLongStringArray",
)
def Restart(self: MccsSubarrayBeam) -> DevVarLongStringArrayType:
"""
Restart the subarray. That is, deconfigure and release all resources.
To modify behaviour for this command, modify the do() method of
the command class.
:return: A tuple containing a result code and the unique ID of the command
"""
handler = self.get_command_object("Restart")
(result_code, message) = handler()
return ([result_code], [message])
[docs] class AbortCommand(SlowCommand[tuple[ResultCode, str]]):
"""A class for MccsSubarrayBeam's Abort() command."""
[docs] def __init__(
self: MccsSubarrayBeam.AbortCommand,
command_tracker: CommandTracker,
component_manager: SubarrayBeamComponentManager,
callback: Callable[[bool], None],
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
states and finishes
:param logger: a logger for this command object to yuse
"""
self._command_tracker = command_tracker
self._component_manager = component_manager
super().__init__(callback=callback, logger=logger)
[docs] def do(
self: MccsSubarrayBeam.AbortCommand,
*args: Any,
**kwargs: Any,
) -> tuple[ResultCode, str]:
"""
Stateless hook for Abort() command functionality.
:param args: positional arguments to the command. This
command does not take any, so this should be empty.
:param kwargs: keyword arguments to the command. This
command does not take any, so this should be empty.
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
command_id = self._command_tracker.new_command(
"Abort", completed_callback=self._completed
)
status, _ = self._component_manager.abort(
functools.partial(self._command_tracker.update_command_info, command_id)
)
# Abort() doesn't get queued, we skip the queue and start immediately in a
# separate thread.
assert status == TaskStatus.IN_PROGRESS
return ResultCode.STARTED, command_id
[docs] @command(dtype_out="DevVarLongStringArray")
def Abort(self: MccsSubarrayBeam) -> DevVarLongStringArrayType:
"""
Abort any long-running command such as ``Configure()`` or ``Scan()``.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("Abort")
(result_code, message) = handler()
return ([result_code], [message])
[docs] class AbortDeviceCommand(SlowCommand[tuple[ResultCode, str]]):
"""A class for MccsSubarray's AbortDevice() command."""
[docs] def __init__(
self: MccsSubarrayBeam.AbortDeviceCommand,
command_tracker: CommandTracker,
component_manager: SubarrayBeamComponentManager,
callback: Callable[[bool], None],
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
states and finishes
:param logger: a logger for this command object to yuse
"""
self._command_tracker = command_tracker
self._component_manager = component_manager
super().__init__(callback=callback, logger=logger)
[docs] def do(
self: MccsSubarrayBeam.AbortDeviceCommand,
*args: Any,
**kwargs: Any,
) -> tuple[ResultCode, str]:
"""
Stateless hook for AbortDevice() command functionality.
:param args: positional arguments to the command. This
command does not take any, so this should be empty.
:param kwargs: keyword arguments to the command. This
command does not take any, so this should be empty.
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
command_id = self._command_tracker.new_command(
"AbortDevice", completed_callback=self._completed
)
status, _ = self._component_manager.abort_device(
functools.partial(self._command_tracker.update_command_info, command_id)
)
assert status == TaskStatus.IN_PROGRESS
return ResultCode.STARTED, command_id
[docs] @command(dtype_out="DevVarLongStringArray")
def AbortDevice(self: MccsSubarrayBeam) -> DevVarLongStringArrayType:
"""
Abort any long-running command such as ``Configure()`` or ``Scan()``.
This will only cancel commands on this device, not further down the hierarchy,
use Abort() for that use case.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("AbortDevice")
(result_code, message) = handler()
return ([result_code], [message])
# ----------
# Run server
# ----------
[docs]def main(*args: str, **kwargs: str) -> int: # pragma: no cover
"""
Entry point for module.
:param args: positional arguments
:param kwargs: named arguments
:return: exit code
"""
return MccsSubarrayBeam.run_server(args=args or None, **kwargs)
if __name__ == "__main__":
main()