connection_manager

frequenz.sdk.microgrid.connection_manager

Microgrid Connection Manager singleton abstraction.

This module provides a singleton abstraction over the microgrid. The main purpose is to provide the connection the microgrid API client and the microgrid component graph.

Classes

frequenz.sdk.microgrid.connection_manager.ConnectionManager

Bases: ABC

Creates and stores core features.

Source code in frequenz/sdk/microgrid/connection_manager.py

class ConnectionManager(ABC):
    """Creates and stores core features."""

    def __init__(self, host: str, port: int) -> None:
        """Create object instance.

        Args:
            host: server host
            port: server port
        """
        super().__init__()
        self._host: str = host
        self._port: int = port

    @property
    def host(self) -> str:
        """Get host of the currently connected server.

        Returns:
            host
        """
        return self._host

    @property
    def port(self) -> int:
        """Get port of the currently connected server.

        Returns:
            port
        """
        return self._port

    @property
    @abstractmethod
    def api_client(self) -> MicrogridApiClient:
        """Get MicrogridApiClient.

        Returns:
            api client
        """

    @property
    @abstractmethod
    def component_graph(self) -> ComponentGraph:
        """Get component graph.

        Returns:
            component graph
        """

    async def _update_api(self, host: str, port: int) -> None:
        self._host = host
        self._port = port

    @abstractmethod
    async def _initialize(self) -> None:
        """Initialize the object. This function should be called only once."""

Attributes

api_client: MicrogridApiClient abstractmethod property

Get MicrogridApiClient.

RETURNS	DESCRIPTION
MicrogridApiClient	api client

component_graph: ComponentGraph abstractmethod property

Get component graph.

RETURNS	DESCRIPTION
ComponentGraph	component graph

host: str property

Get host of the currently connected server.

RETURNS	DESCRIPTION
str	host

port: int property

Get port of the currently connected server.

RETURNS	DESCRIPTION
int	port

Functions

__init__(host, port)

Create object instance.

PARAMETER	DESCRIPTION
host	server host TYPE: str
port	server port TYPE: int

Source code in frequenz/sdk/microgrid/connection_manager.py

def __init__(self, host: str, port: int) -> None:
    """Create object instance.

    Args:
        host: server host
        port: server port
    """
    super().__init__()
    self._host: str = host
    self._port: int = port

Functions

frequenz.sdk.microgrid.connection_manager.get()

Get the MicrogridApi instance created by initialize().

This function should be only called after initialize().

RAISES	DESCRIPTION
RuntimeError	Raised when: * If initialize() method was not called before this call. * If initialize() methods was called but was not awaited and instance was not created yet.

RETURNS	DESCRIPTION
ConnectionManager	MicrogridApi instance.

Source code in frequenz/sdk/microgrid/connection_manager.py

def get() -> ConnectionManager:
    """Get the MicrogridApi instance created by initialize().

    This function should be only called after initialize().

    Raises:
        RuntimeError: Raised when:
            * If `initialize()` method was not called before this call.
            * If `initialize()` methods was called but was not awaited and instance was
                not created yet.

    Returns:
        MicrogridApi instance.
    """
    if _CONNECTION_MANAGER is None:
        raise RuntimeError(
            "ConnectionManager is not initialized. "
            "Call `await microgrid.initialize()` first."
        )

    return _CONNECTION_MANAGER

frequenz.sdk.microgrid.connection_manager.initialize(host, port) async

Initialize the MicrogridApi. This function should be called only once.

PARAMETER	DESCRIPTION
host	Microgrid host TYPE: str
port	Microgrid port TYPE: int

RAISES	DESCRIPTION
AssertionError	If method was called more then once.

Source code in frequenz/sdk/microgrid/connection_manager.py

async def initialize(host: str, port: int) -> None:
    """Initialize the MicrogridApi. This function should be called only once.

    Args:
        host: Microgrid host
        port: Microgrid port

    Raises:
        AssertionError: If method was called more then once.
    """
    # From Doc: pylint just try to discourage this usage.
    # That doesn't mean you cannot use it.
    global _CONNECTION_MANAGER  # pylint: disable=global-statement

    if _CONNECTION_MANAGER is not None:
        raise AssertionError("MicrogridApi was already initialized.")

    _logger.info("Connecting to microgrid at %s:%s", host, port)

    microgrid_api = _InsecureConnectionManager(host, port)
    await microgrid_api._initialize()  # pylint: disable=protected-access

    # Check again that _MICROGRID_API is None in case somebody had the great idea of
    # calling initialize() twice and in parallel.
    if _CONNECTION_MANAGER is not None:
        raise AssertionError("MicrogridApi was already initialized.")

    _CONNECTION_MANAGER = microgrid_api