123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263 |
- # 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 aiohttp import web
- from pyopenadr.service import EventService, PollService, RegistrationService, ReportService, OptService
- class OpenADRServer:
- _MAP = {'on_created_event': EventService,
- 'on_request_event': EventService,
- 'on_register_report': ReportService,
- 'on_create_report': ReportService,
- 'on_created_report': ReportService,
- 'on_request_report': ReportService,
- 'on_update_report': ReportService,
- 'on_poll': PollService,
- 'on_query_registration': RegistrationService,
- 'on_create_party_registration': RegistrationService,
- 'on_cancel_party_registration': RegistrationService}
- def __init__(self, vtn_id):
- self.app = web.Application()
- self.services = {'event_service': EventService(vtn_id),
- 'report_service': ReportService(vtn_id),
- 'poll_service': PollService(vtn_id),
- 'opt_service': OptService(vtn_id),
- 'registration_service': RegistrationService(vtn_id)}
- self.app.add_routes([web.post(f"/OpenADR2/Simple/2.0b/{s.__service_name__}", s.handler) for s in self.services.values()])
- self.__setattr__ = self.add_handler
- def run(self):
- """
- Starts the asyncio-loop and runs the server in it. This function is
- blocking. For other ways to run the server in a more flexible context,
- please refer to the `aiohttp documentation
- <https://docs.aiohttp.org/en/stable/web_advanced.html#aiohttp-web-app-runners>`_.
- """
- web.run_app(self.app)
- def add_handler(self, name, func):
- """
- Add a handler to the OpenADRServer.
- """
- print("Called add_handler", name, func)
- if name in self._MAP:
- setattr(self._MAP[name], name, staticmethod(func))
- else:
- raise NameError(f"Unknown handler {name}. Correct handler names are: {self._MAP.keys()}")
|