interfaces.py

import abc
import datetime
import functools
from typing import Any, Collection, Mapping, Optional

from .types import *

__all__ = [
    "FleafSemanticsDescriptor", "FleafBaseAPI", "FleafForecastingExtension",
    "ContinuousAvailabilityFleafAPI", "SessionBasedFleafAPI", "AllSemanticsFleafAPI",
]


class FleafSemanticsDescriptor(abc.ABC):

    @classmethod
    @abc.abstractmethod
    def _get_semantics_type(cls) -> Optional[str]:
        return None


class FleafBaseAPI(FleafSemanticsDescriptor):
    """Defines the basic methods of an API supporting the Fleaf protocol.

    This class and its subclasses can be used in MixIn-styled
    inheritance, and should therefore probably be inherited from before
    any other non-mixin classes.
    """

    def __init__(self, *args, **kwargs):
        self._supported_semantics = set()
        for cls in type(self).__mro__:
            if issubclass(cls, FleafSemanticsDescriptor):
                self._supported_semantics.add(cls._get_semantics_type())
        try:
            self._supported_semantics.remove(None)
        except KeyError:
            pass
        # MixIn style: Initialize any other base classes defined by a child class.
        super().__init__(*args, **kwargs)

    @property
    def supported_semantics(self) -> Collection[str]:
        """A collection of architecture styles that are supported by this API
        implementation.

        Clients should select a style they understand and prefer for
        communication.
        """
        return self._supported_semantics

    @property
    @abc.abstractmethod
    def supported_traits(self) -> Collection[FleafRequestableTrait]:
        """A collection of traits that can be requested from this API
        implementation."""
        return []

    @abc.abstractmethod
    def grid_id(self, consumer_id: Optional[str]) -> str:
        """Return an identifier for the microgrid the consumer is part of.

        If the API instance is only responsible for one microgrid,
        and no consumer traits are requested, `None` may be passed as the consumer ID.
        An API implementation may reject an ID of `None` if it cannot process it.
        """
        return ""


class FleafForecastingExtension(FleafSemanticsDescriptor):
    SEMANTICS = "with forecasts"

    @classmethod
    def _get_semantics_type(cls):
        return __class__.SEMANTICS

    @property
    @abc.abstractmethod
    def forecastable_traits(self) -> Collection[FleafRequestableTrait]:
        """A collection of traits that can be forecasted by this API
        implementation."""
        return []

    @abc.abstractmethod
    def forecast(self, consumer_id: Optional[str], trait: FleafRequestableTrait, duration: datetime.timedelta,
                 series_properties: Optional[TimeSeriesProperties] = None) -> TimeSeries[Any]:
        """Return a forecast of the given trait over the given duration.

        If the API instance is only responsible for one microgrid,
        and no consumer traits are requested, `None` may be passed as the consumer ID.
        An API implementation may reject an ID of `None` if it cannot process it.

        `forecastable_traits` returns a collection of traits for which
        this API implementation is able to return a forecast.

        The time series spans as much of the given time window as
        possible.
        """
        return []


class ContinuousAvailabilityFleafAPI(FleafBaseAPI, FleafForecastingExtension):
    SEMANTICS = "continuous availability"

    @classmethod
    def _get_semantics_type(cls):
        return __class__.SEMANTICS

    @abc.abstractmethod
    def series(self, consumer_id: Optional[str], trait: FleafRequestableTrait,
               start_time: datetime.datetime, end_time: datetime.datetime,
               series_properties: Optional[TimeSeriesProperties] = None) -> TimeSeries[Any]:
        """Return a time series of the given trait that spans as much of the
        given time window as possible.

        If the API instance is only responsible for one microgrid,
        and no consumer traits are requested, `None` may be passed as the consumer ID.
        An API implementation may reject an ID of `None` if it cannot process it.
        """
        pass

    def historical(self, consumer_id: Optional[str], trait: FleafRequestableTrait, duration: datetime.timedelta,
                   series_properties: Optional[TimeSeriesProperties] = None) -> TimeSeries[Any]:
        """Return a historical time series of the given trait over the given
        duration.

        If the API instance is only responsible for one microgrid,
        and no consumer traits are requested, `None` may be passed as the consumer ID.
        An API implementation may reject an ID of `None` if it cannot process it.

        The time series spans as much of the given time window as
        possible.
        """
        end = datetime.datetime.now().astimezone()
        start = end - duration
        return self.series(consumer_id, trait, start, end, series_properties)

    def instant(self, consumer_id: Optional[str], trait: FleafRequestableTrait) -> Any:
        """Return the given trait at the current time instant.

        If the API instance is only responsible for one microgrid,
        and no consumer traits are requested, `None` may be passed as the consumer ID.
        An API implementation may reject an ID of `None` if it cannot process it.
        """
        start = datetime.datetime.now().astimezone()
        resolution = datetime.timedelta(seconds=1)
        end = start + resolution
        result = self.series(consumer_id, trait, start, end, series_properties=TimeSeriesProperties(resolution))
        if isinstance(result, Mapping):
            return next(iter(result.values()))
        else:
            return result[0][1]

    def forecast(self, consumer_id: Optional[str], trait: FleafRequestableTrait, duration: datetime.timedelta,
                 series_properties: Optional[TimeSeriesProperties] = None) -> TimeSeries[Any]:
        start = datetime.datetime.now().astimezone()
        end = start + duration
        return self.series(consumer_id, trait, start, end, series_properties)


IndexedTimeSeriesCollection = Mapping[FleafRequestableTrait, TimeSeries[Any]]


class SessionBasedFleafAPI(FleafBaseAPI):
    SEMANTICS = "session based"

    @classmethod
    def _get_semantics_type(cls):
        return __class__.SEMANTICS

    @abc.abstractmethod
    def start_measurement(self, consumer_id: Optional[str]) -> str:
        """Return a session ID that can be used to make further inquiries about
        the energy system.

        If the API instance is only responsible for one microgrid,
        and no consumer traits are later requested, `None` may be passed as the consumer ID.
        An API implementation may reject an ID of `None` if it cannot process it.
        """
        pass

    @abc.abstractmethod
    def measurement_status(self, session_id: str, traits: Collection[FleafRequestableTrait],
                           series_properties: Optional[TimeSeriesProperties] = None) -> IndexedTimeSeriesCollection:
        """Return the requested traits over the duration of the given
        measurement session."""
        pass

    def end_measurement(self, session_id: str, *status_args, **status_kwargs) -> IndexedTimeSeriesCollection:
        """End the given measurement session and return the requested traits
        over the duration of said session."""
        result = self.measurement_status(session_id, *status_args, **status_kwargs)
        self.unregister_session(session_id)
        return result

    @abc.abstractmethod
    def unregister_session(self, session_id: str) -> None:
        """Mark the given session as expired."""
        pass


class AllSemanticsFleafAPI(ContinuousAvailabilityFleafAPI, SessionBasedFleafAPI):
    pass