Async base controller

Async base controller class for odin-control adapters.

This module defines the base error class and abstract controller interface for odin control system adapters.

Tim Nicholls, STFC Detector Systems Software Group

`AsyncBaseController`

Bases: ABC

Abstract base class for asynchronous controllers.

This class defines the interface that all async adapter controllers must implement to provide consistent behavior across different adapter types. Controllers handle the core logic and device communication for adapters.

At a minimum, derived controller classes must implement the following methods: - init(options) - get(path, with_metadata=False)

The following methods can be optionally implemented to provide additional functionality: - initialize(adapters) - cleanup() - set(path, data) - create(path, data) - delete(path)

Source code in src/odin_control/adapters/async_base_controller.py

class AsyncBaseController(ABC):
    """Abstract base class for asynchronous controllers.

    This class defines the interface that all async adapter controllers must implement to provide
    consistent behavior across different adapter types. Controllers handle the core logic and device
    communication for adapters.

    At a minimum, derived controller classes must implement the following methods:
    - __init__(options)
    - get(path, with_metadata=False)

    The following methods can be optionally implemented to provide additional functionality:
    - initialize(adapters)
    - cleanup()
    - set(path, data)
    - create(path, data)
    - delete(path)
    """

    @abstractmethod
    def __init__(self, options):
        """Initialize the asynchronous controller with configuration options.

        :param options: Dictionary containing configuration options for the controller.
        """
        ...  # pragma: no cover

    def __await__(self):
        """Make the controller awaitable to resolve async attributes."""
        async def closure():
            """Await all async attributes of the controller."""
            awaitable_attrs = [attr for attr in self.__dict__.values() if inspect.isawaitable(attr)]
            await asyncio.gather(*awaitable_attrs)
            return self

        return closure().__await__()

    async def initialize(self, adapters: Dict[str, object]) -> None:
        """Initialize the asynchronous controller with access to other adapters.

        This method is called after all adapters have been loaded and allows
        the controller to establish inter-adapter communication if needed.

        :param adapters: Dictionary mapping adapter names to adapter instances.
        """
        self.adapters = adapters

    async def cleanup(self) -> None:
        """Clean up controller resources and state.

        This method should be implemented to properly clean up any resources,
        close connections, or perform other shutdown tasks when the controller
        is being stopped or destroyed.
        """
        ...

    @abstractmethod
    async def get(self, path: str, with_metadata: bool = False) -> Dict[str, Any]:
        """Asynchronously get data from the controller at the specified path.

        :param path: The path string identifying the resource to retrieve.
        :param with_metadata: Whether to include metadata in the response.
        :return: A dictionary containing the requested data.
        """
        ...  # pragma: no cover

    async def set(self, path: str, data: Dict[str, Any]) -> None:
        """Asynchronously set data in the controller at the specified path.

        :param path: The path string identifying the resource to modify.
        :param data: A dictionary containing the data to set.
        """
        raise NotImplementedError()

    async def create(self, path: str, data: Dict[str, Any]) -> Dict[str, Any]:
        """Asynchronously create a new resource at the specified path.

        :param path: The path string identifying where to create the resource.
        :param data: A dictionary containing the data for the new resource.
        :return: A dictionary containing the created resource's data.
        """
        raise NotImplementedError()

    async def delete(self, path: str) -> None:
        """Asynchronously delete a resource at the specified path.

        :param path: The path string identifying the resource to delete.
        """
        raise NotImplementedError()

`await()`

Make the controller awaitable to resolve async attributes.

Source code in src/odin_control/adapters/async_base_controller.py

def __await__(self):
    """Make the controller awaitable to resolve async attributes."""
    async def closure():
        """Await all async attributes of the controller."""
        awaitable_attrs = [attr for attr in self.__dict__.values() if inspect.isawaitable(attr)]
        await asyncio.gather(*awaitable_attrs)
        return self

    return closure().__await__()

`init(options)` `abstractmethod`

Initialize the asynchronous controller with configuration options.

Parameters:

Name	Type	Description	Default
`options`		Dictionary containing configuration options for the controller.	required

Source code in src/odin_control/adapters/async_base_controller.py

@abstractmethod
def __init__(self, options):
    """Initialize the asynchronous controller with configuration options.

    :param options: Dictionary containing configuration options for the controller.
    """
    ...  # pragma: no cover

`cleanup()` `async`

Clean up controller resources and state.

This method should be implemented to properly clean up any resources, close connections, or perform other shutdown tasks when the controller is being stopped or destroyed.

Source code in src/odin_control/adapters/async_base_controller.py

async def cleanup(self) -> None:
    """Clean up controller resources and state.

    This method should be implemented to properly clean up any resources,
    close connections, or perform other shutdown tasks when the controller
    is being stopped or destroyed.
    """
    ...

`create(path, data)` `async`

Asynchronously create a new resource at the specified path.

Parameters:

Name	Type	Description	Default
`path`	`str`	The path string identifying where to create the resource.	required
`data`	`Dict[str, Any]`	A dictionary containing the data for the new resource.	required

Returns:

Type	Description
`Dict[str, Any]`	A dictionary containing the created resource's data.

Source code in src/odin_control/adapters/async_base_controller.py

async def create(self, path: str, data: Dict[str, Any]) -> Dict[str, Any]:
    """Asynchronously create a new resource at the specified path.

    :param path: The path string identifying where to create the resource.
    :param data: A dictionary containing the data for the new resource.
    :return: A dictionary containing the created resource's data.
    """
    raise NotImplementedError()

`delete(path)` `async`

Asynchronously delete a resource at the specified path.

Parameters:

Name	Type	Description	Default
`path`	`str`	The path string identifying the resource to delete.	required

Source code in src/odin_control/adapters/async_base_controller.py

async def delete(self, path: str) -> None:
    """Asynchronously delete a resource at the specified path.

    :param path: The path string identifying the resource to delete.
    """
    raise NotImplementedError()

`get(path, with_metadata=False)` `abstractmethod` `async`

Asynchronously get data from the controller at the specified path.

Parameters:

Name	Type	Description	Default
`path`	`str`	The path string identifying the resource to retrieve.	required
`with_metadata`	`bool`	Whether to include metadata in the response.	`False`

Returns:

Type	Description
`Dict[str, Any]`	A dictionary containing the requested data.

Source code in src/odin_control/adapters/async_base_controller.py

@abstractmethod
async def get(self, path: str, with_metadata: bool = False) -> Dict[str, Any]:
    """Asynchronously get data from the controller at the specified path.

    :param path: The path string identifying the resource to retrieve.
    :param with_metadata: Whether to include metadata in the response.
    :return: A dictionary containing the requested data.
    """
    ...  # pragma: no cover

`initialize(adapters)` `async`

Initialize the asynchronous controller with access to other adapters.

This method is called after all adapters have been loaded and allows the controller to establish inter-adapter communication if needed.

Parameters:

Name	Type	Description	Default
`adapters`	`Dict[str, object]`	Dictionary mapping adapter names to adapter instances.	required

Source code in src/odin_control/adapters/async_base_controller.py

async def initialize(self, adapters: Dict[str, object]) -> None:
    """Initialize the asynchronous controller with access to other adapters.

    This method is called after all adapters have been loaded and allows
    the controller to establish inter-adapter communication if needed.

    :param adapters: Dictionary mapping adapter names to adapter instances.
    """
    self.adapters = adapters

`set(path, data)` `async`

Asynchronously set data in the controller at the specified path.

Parameters:

Name	Type	Description	Default
`path`	`str`	The path string identifying the resource to modify.	required
`data`	`Dict[str, Any]`	A dictionary containing the data to set.	required

Source code in src/odin_control/adapters/async_base_controller.py

async def set(self, path: str, data: Dict[str, Any]) -> None:
    """Asynchronously set data in the controller at the specified path.

    :param path: The path string identifying the resource to modify.
    :param data: A dictionary containing the data to set.
    """
    raise NotImplementedError()

Async base controller