# -*- 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 component management for the MCCS controller."""
from __future__ import annotations
from typing import Hashable, Iterable, Mapping, cast
from ska_low_mccs_common.resource_manager import ResourcePool, _ReadyResourceManager
__all__ = ["ControllerResourceManager"]
[docs]class ControllerResourceManager:
"""A resource manager for the controller component manager."""
[docs] def __init__(
self: ControllerResourceManager,
subarrays: Iterable[str],
subarray_beams: Iterable[str],
station_beams: Iterable[str],
channel_blocks: Iterable[int],
) -> None:
"""
Initialise a new instance.
:param subarrays: all subarrays to be managed by this resource
manager
:param subarray_beams: all subarray beams to be managed by this
resource manager
:param station_beams: all station beams to be managed by this
resource manager (as a resource pool)
:param channel_blocks: all channel blocks to be managed by this
resource manager
"""
self._resource_manager = _ReadyResourceManager(
subarrays,
subarray_beams=subarray_beams,
station_beams=station_beams,
channel_blocks=channel_blocks,
)
self.resource_pool = ResourcePool(
station_beams=station_beams,
)
[docs] def set_ready(
self: ControllerResourceManager,
subarray: str,
is_ready: bool,
) -> None:
"""
Set the health of a resource.
:param subarray: the subarray to be set as ready
:param is_ready: whether the subarray is ready or not
"""
self._resource_manager.set_ready(subarray, is_ready)
[docs] def allocate(
self: ControllerResourceManager,
subarray: str,
**resources: Iterable[Hashable],
) -> None:
"""
Allocate resources to a subarray.
:param subarray: the subarray to which resources are to be
allocated
:param resources: the resources to allocate. Each keyword
specifies a resource type, with the value a list of the
resources of that type to be allocated. For example:
.. code-block:: python
controller_resource_manager.allocate(
"low-mccs/subarray/01",
stations=[
["low-mccs/station/001", "low-mccs/station/002"]
],
station_beams=[["low-mccs/beam/01", "low-mccs/beam/02"]]
channel_blocks=[2, 3],
)
"""
# scrape stations from iterable - these are not a resource
# stations can be shared between subarrays
if "stations" in resources:
resources.pop("stations", None)
self._resource_manager.allocate(subarray, **resources)
[docs] def deallocate(
self: ControllerResourceManager,
**resources: Iterable[Hashable],
) -> None:
"""
Deallocate resources (regardless of what subarray they are allocated to.
:param resources: the resources to deallocate. Each keyword
specifies a resource type, with the value a list of the
resources of that type to be deallocated. For example:
.. code-block:: python
controller_resource_manager.deallocate(
stations=[
"low-mccs/station/001", "low-mccs/station/002"
],
channel_blocks=[2, 3],
)
"""
self._resource_manager.deallocate(**resources)
[docs] def deallocate_from(
self: ControllerResourceManager,
subarray: str,
) -> None:
"""
Deallocate all resources from a subarray.
:param subarray: the subarray to which resources are to be
allocated
"""
self._resource_manager.deallocate_from(subarray)
[docs] def get_allocated(
self: ControllerResourceManager,
subarray: str,
) -> Mapping[str, Iterable[str]]:
"""
Return the resources allocated to a given subarray.
:param subarray: TRL of the subarray for which the allocated
resources are to be returned
:return: the resources allocated to the subarray.
"""
return cast(
Mapping[str, Iterable[str]],
self._resource_manager.get_allocated(subarray),
)