Skip to content

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) -> ApiClient:
        """Get ApiClient.

        Returns:
            api client
        """

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

        Returns:
            component graph
        """

    @property
    @abstractmethod
    def microgrid_id(self) -> int | None:
        """Get the ID of the microgrid if available.

        Returns:
            the ID of the microgrid if available, None otherwise.
        """

    @property
    @abstractmethod
    def location(self) -> Location | None:
        """Get the location of the microgrid if available.

        Returns:
            the location of the microgrid if available, None otherwise.
        """

    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 abstractmethod property ¤
api_client: ApiClient

Get ApiClient.

RETURNS DESCRIPTION
ApiClient

api client

component_graph abstractmethod property ¤
component_graph: ComponentGraph

Get component graph.

RETURNS DESCRIPTION
ComponentGraph

component graph

host property ¤
host: str

Get host of the currently connected server.

RETURNS DESCRIPTION
str

host

location abstractmethod property ¤
location: Location | None

Get the location of the microgrid if available.

RETURNS DESCRIPTION
Location | None

the location of the microgrid if available, None otherwise.

microgrid_id abstractmethod property ¤
microgrid_id: int | None

Get the ID of the microgrid if available.

RETURNS DESCRIPTION
int | None

the ID of the microgrid if available, None otherwise.

port property ¤
port: int

Get port of the currently connected server.

RETURNS DESCRIPTION
int

port

Functions¤
__init__ ¤
__init__(host: str, port: int) -> None

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 async ¤

initialize(host: str, port: int) -> None

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