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, server_url: str) -> None:
        """Create object instance.

        Args:
            server_url: The location of the microgrid API server in the form of a URL.
                The following format is expected: `grpc://hostname{:port}{?ssl=ssl}`,
                where the port should be an int between `0` and `65535` (defaulting to
                `9090`) and ssl should be a boolean (defaulting to false). For example:
                `grpc://localhost:1090?ssl=true`.
        """
        super().__init__()
        self._server_url = server_url

    @property
    def server_url(self) -> str:
        """The location of the microgrid API server in the form of a URL."""
        return self._server_url

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

        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, server_url: str) -> None:
        self._server_url = server_url

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

Attributes

api_client abstractmethod property

api_client: MicrogridApiClient

Get the MicrogridApiClient.

RETURNS	DESCRIPTION
MicrogridApiClient	api client

component_graph abstractmethod property

component_graph: ComponentGraph

Get component graph.

RETURNS	DESCRIPTION
ComponentGraph	component graph

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.

server_url property

server_url: str

The location of the microgrid API server in the form of a URL.

Functions

__init__

__init__(server_url: str) -> None

Create object instance.

PARAMETER	DESCRIPTION
server_url	The location of the microgrid API server in the form of a URL. The following format is expected: grpc://hostname{:port}{?ssl=ssl}, where the port should be an int between 0 and 65535 (defaulting to 9090) and ssl should be a boolean (defaulting to false). For example: grpc://localhost:1090?ssl=true. TYPE: str

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

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

    Args:
        server_url: The location of the microgrid API server in the form of a URL.
            The following format is expected: `grpc://hostname{:port}{?ssl=ssl}`,
            where the port should be an int between `0` and `65535` (defaulting to
            `9090`) and ssl should be a boolean (defaulting to false). For example:
            `grpc://localhost:1090?ssl=true`.
    """
    super().__init__()
    self._server_url = server_url

Functions

frequenz.sdk.microgrid.connection_manager.get

get() -> ConnectionManager

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(server_url: str) -> None

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

PARAMETER	DESCRIPTION
server_url	The location of the microgrid API server in the form of a URL. The following format is expected: grpc://hostname{:port}{?ssl=ssl}, where the port should be an int between 0 and 65535 (defaulting to 9090) and ssl should be a boolean (defaulting to false). For example: grpc://localhost:1090?ssl=true. TYPE: str

RAISES	DESCRIPTION
AssertionError	If method was called more then once.

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

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

    Args:
        server_url: The location of the microgrid API server in the form of a URL.
            The following format is expected: `grpc://hostname{:port}{?ssl=ssl}`,
            where the port should be an int between `0` and `65535` (defaulting to
            `9090`) and ssl should be a boolean (defaulting to false). For example:
            `grpc://localhost:1090?ssl=true`.

    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", server_url)

    microgrid_api = _InsecureConnectionManager(server_url)
    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