pyopenadr/openleadr/service/event_service.py

# SPDX-License-Identifier: Apache-2.0

# Copyright 2020 Contributors to OpenLEADR

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from . import service, handler, VTNService
from asyncio import iscoroutine
from .. import objects
import logging
logger = logging.getLogger('openleadr')


@service('EiEvent')
class EventService(VTNService):

    def __init__(self, vtn_id, polling_method='internal', message_queues=None):
        super().__init__(vtn_id)
        self.polling_method = polling_method
        self.message_queues = message_queues
        self.pending_events = {}        # Holds the event callbacks

    @handler('oadrRequestEvent')
    async def request_event(self, payload):
        """
        The VEN requests us to send any events we have.
        """
        result = self.on_request_event(payload['ven_id'])
        if iscoroutine(result):
            result = await result
        if result is None:
            return 'oadrDistributeEvent', {'events': []}
        if isinstance(result, dict):
            return 'oadrDistributeEvent', {'events': [result]}
        if isinstance(result, objects.Event):
            return 'oadrDistributeEvent', {'events': [result]}
        if isinstance(result, list):
            return 'oadrDistributeEvent', {'events': result}
        else:
            raise TypeError("Event handler should return None, a dict or a list")

    def on_request_event(self, ven_id):
        """
        Placeholder for the on_request_event handler.
        """
        logger.warning("You should implement and register your own on_request_event handler "
                       "that returns the next event for a VEN. This handler will receive a "
                       "ven_id as its only argument, and should return None (if no events are "
                       "available), a single Event, or a list of Events.")
        return None

    @handler('oadrCreatedEvent')
    async def created_event(self, payload):
        """
        The VEN informs us that they created an EiEvent.
        """
        if self.polling_method == 'internal':
            for event_response in payload['event_responses']:
                if event_response['event_id'] in self.pending_events:
                    ven_id = payload['ven_id']
                    event_id = event_response['event_id']
                    opt_type = event_response['opt_type']
                    callback = self.pending_events.pop(event_id)
                    result = callback(ven_id=ven_id, event_id=event_id, opt_type=opt_type)
                    if iscoroutine(result):
                        result = await result
        else:
            result = self.on_created_event(ven_id=ven_id, event_id=event_id, opt_type=opt_type)
            if iscoroutine(result):
                result = await(result)
        return 'oadrResponse', {}

    def on_created_event(self, ven_id, event_id, opt_type):
        """
        Placeholder for the on_created_event handler.
        """
        logger.warning("You should implement and register you own on_created_event handler "
                       "to receive the opt status for an Event that you sent to the VEN. This "
                       "handler will receive a ven_id, event_id and opt_status. "
                       "You don't need to return anything from this handler.")
        return None