Server¶
If you are implementing an OpenADR Server (“Virtual Top Node”) using OpenLEADR, read this page.
1-minute VTN example¶
Here’s an example of a server that accepts registrations from a VEN named ‘ven_123’, requests all reports that it offers, and creates an Event for this VEN.
import asyncio
from datetime import datetime, timezone, timedelta
from openleadr import OpenADRServer, enable_default_logging
from functools import partial
enable_default_logging()
async def on_create_party_registration(registration_info):
"""
Inspect the registration info and return a ven_id and registration_id.
"""
if registration_info['ven_name'] == 'ven123':
ven_id = 'ven_id_123'
registration_id = 'reg_id_123'
return ven_id, registration_id
else:
return False
async def on_register_report(ven_id, resource_id, measurement, unit, scale,
min_sampling_interval, max_sampling_interval):
"""
Inspect a report offering from the VEN and return a callback and sampling interval for receiving the reports.
"""
callback = partial(on_update_report, ven_id=ven_id, resource_id=resource_id, measurement=measurement)
sampling_interval = min_sampling_interval
return callback, sampling_interval
async def on_update_report(data, ven_id, resource_id, measurement):
"""
Callback that receives report data from the VEN and handles it.
"""
for time, value in data:
print(f"Ven {ven_id} reported {measurement} = {value} at time {time} for resource {resource_id}")
async def event_response_callback(ven_id, event_id, opt_type):
"""
Callback that receives the response from a VEN to an Event.
"""
print(f"VEN {ven_id} responded to Event {event_id} with: {opt_type}")
# Create the server object
server = OpenADRServer(vtn_id='myvtn')
# Add the handler for client (VEN) registrations
server.add_handler('on_create_party_registration', on_create_party_registration)
# Add the handler for report registrations from the VEN
server.add_handler('on_register_report', on_register_report)
# Add a prepared event for a VEN that will be picked up when it polls for new messages.
server.add_event(ven_id='ven_id_123',
signal_name='simple',
signal_type='level',
intervals=[{'dtstart': datetime(2021, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
'duration': timedelta(minutes=10),
'signal_payload': 1}],
callback=event_response_callback)
# Run the server on the asyncio event loop
loop = asyncio.get_event_loop()
loop.create_task(server.run())
loop.run_forever()
Read on for more details!
Registration¶
If a client (VEN) wants to register for the first time, it will go through a Registration procedure.
Implementation Checklist
Create a handler that decides what to do with new registrations, based on their registration info.
The client will send a oadrQueryRegistration message. The server will respond with a oadrCreatedPartyRegistration message containing a list of its capabilities, notably the implemented OpenADR protocol versions and the available Transport Mechanisms (HTTP and/or XMPP).
The client will then usually send a oadrCreatePartyRegistration message, in which it registers to a specific OpenADR version and Transport Method. The server must then decide what it wants to do with this registration.
In the case that the registration is accepted, the VTN will generate a venID and a RegistrationID for this VEN and respond with a oadrCreatedPartyRegistration message.
In your application, when a VEN sends a oadrCreatePartyRegistration request, it will call your on_create_party_registration handler. This handler must somehow look up what to do with this request, and respond with a ven_id, registration_id tuple.
Example implementation:
from openleadr.utils import generate_id
async def on_create_party_registration(payload):
ven_name = payload['ven_name']
# Check whether or not this VEN is allowed to register
result = await database.query("""SELECT COUNT(*)
FROM vens
WHERE ven_name = ?""",
(payload['ven_name'],))
if result == 1:
# Generate an ID for this registration
ven_id = generate_id()
registration_id = generate_id()
# Store the registration in a database (pseudo-code)
await database.query("""UPDATE vens
SET ven_id = ?
registration_id = ?
WHERE ven_name = ?""",
(ven_id, registration_id, ven_name))
# Return the registration ID.
# This will be put into the correct form by the OpenADRServer.
return ven_id, registration_id
Events¶
The server (VTN) is expected to know when it needs to inform the clients (VENs) of certain events that they must respond to. This could be a predicted shortage or overage of available power in a certain electricity grid area, for example.
The easiest way to supply events to a VEN is by using OpenLEADR’s built-in message queing system. You simply add an event for a ven using the server.add_event method. You supply the ven_id for which the event is required, as well as the signal_name, signal_type, intervals and targets. This will build an event object with a single signal for a VEN. If you need more flexibility, you can alternatively construct the event dictionary yourself and supply it directly to the add_raw_event method.
The VEN can decide whether to opt in or opt out of the event. To be notified of their opt status, you supply a callback handler which will be called when the VEN has responded to the event request.
from openleadr import OpenADRServer
from functools import partial
from datetime import datetime, timezzone
async def event_callback(ven_id, event_id, opt_status):
print(f"VEN {ven_id} responded {opt_status} to event {event_id}")
server = OpenADRServer(vtn_id='myvtn')
event_id = server.add_event(ven_id='ven123',
signal_name='simple',
signal_type='level',
intervals=[{'dtstart': datetime(2020. 1, 1, 12, 0, 0, tzinfo=timezone.utc),
'signal_payload': 1},
{'dtstart': datetime(2020. 1, 1, 12, 15, 0, tzinfo=timezone.utc),
'signal_payload': 0}],
target=[{'resource_id': 'Device001'}],
callback=event_callback)
Alternatively, you can use the handy constructors in openleadr.objects to format parts of the event:
from openleadr import OpenADRServer
from openleadr.objects import Target, Interval
from datetime import datetime, timezone
from functools import partial
async def event_callback(ven_id, event_id, opt_status):
print(f"VEN {ven_id} responded {opt_status} to event {event_id}")
server = OpenADRServer(vtn_id='myvtn')
event_id = server.add_event(ven_id='ven123',
signal_name='simple',
signal_type='level',
intervals=[Interval(dtstart=datetime(2020, 1, 1, 12, 15, 0, tzinfo=timezone.utc),
signal_payload=0),
Interval(dtstart=datetime(2020, 1, 1, 12, 15, 0, tzinfo=timezone.utc),
signal_payload=1)]
target=[Target(resource_id='Device001')],
callback=event_callback)
If you want to add a “raw” event directly, you can use this example as a guid:
from openleadr import OpenADRServer
from openleadr.objects import Event, EventDescriptor, EventSignal, Target, Interval
from datetime import datetime, timezone
from functools import partial
async def event_callback(ven_id, event_id, opt_status):
print(f"VEN {ven_id} responded {opt_status} to event {event_id}")
server = OpenADRServer(vtn_id='myvtn')
now = datetime.now()
event = Event(event_descriptor=EventDescriptor(event_id='event001',
modification_number=0,
event_status='far',
market_context='http://marketcontext01'),
event_signals=[EventSignal(signal_id='signal001',
signal_type='level',
signal_name='simple',
intervals=[Interval(dtstart=now,
duration=timedelta(minutes=10),
signal_payload=1)]),
EventSignal(signal_id='signal002',
signal_type='price',
signal_name='ELECTRICITY_PRICE',
intervals=[Interval(dtstart=now,
duration=timedelta(minutes=10),
signal_payload=1)])],
targets=[Target(ven_id='ven123')])
server.add_raw_event(ven_id='ven123', event=event, callback=event_callback)
If you want to add an event and wait for the response in a single coroutine, you can pass an asyncio Future instead of a function or coroutine as the callback argument:
import asyncio
...
async def generate_event():
loop = asyncio.get_event_loop()
opt_status_future = loop.create_future()
server.add_event(..., callback=opt_status_future)
opt_status = await opt_status_future
print(f"The opt status for this event is {opt_status}")
A word on event targets¶
The Target of your Event is an indication for the VEN which resources or devices should be affected. You can supply the target of the event in serveral ways:
Assigning the
targetparameter with a singleobjects.Targetobject.Assigning the
targetsparameter with a list ofobjects.Targetobjects.Assigning the
targets_by_typeparameters with a dict, that lists targets grouped by their type, like this:
server.add_event(...
targets_by_type={'resource_id': ['resource01', 'resource02'],
'group_id': ['group01', 'group02']}
)
If you dont assign any Target, the target will be set to the ven_id that you specified.
Reports¶
Please see the Reporting section.
Things you should implement¶
You should implement the following handlers:
on_create_party_registration(registration_info)on_register_report(ven_id, resource_id, measurement, unit, scale, min_sampling_interval, max_sampling_interval)ven_lookup(ven_id): a function that returns a dict with the ‘ven_name’,'ven_id','fingerprint'and'registration_id'for the givenven_id(see below.). This is used to automatically reject requests from VENs that the VTN does not know, and to authenticate the VENs message signatures.
Optionally:
on_poll(ven_id); only if you don’t want to use the internal message queue.ven_lookup(ven_id): a function or coroutine that openleadr can use to check if we know a VEN. Signature:
Signing Messages¶
The OpenLEADR can sign your messages and validate incoming messages. For some background, see the Message Signing.
Example implementation:
from openleadr import OpenADRServr
def ven_lookup(ven_id):
# Look up the information about this VEN.
ven_info = database.lookup('vens').where(ven_id=ven_id) # Pseudo code
if ven_info:
return {'ven_id': ven_info['ven_id'],
'ven_name': ven_info['ven_name'],
'fingerprint': ven_info['fingerprint'],
'registration_id': ven_info['registration_id']}
else:
return {}
server = OpenADRServer(vtn_id='MyVTN',
cert='/path/to/cert.pem',
key='/path/to/private/key.pem',
passphrase='mypassphrase',
fingerprint_lookup=fingerprint_lookup)
The VEN’s fingerprint should be obtained from the VEN outside of OpenADR.
Message Handlers¶
Your server has to deal with the different OpenADR messages. The way this works is that OpenLEADR will expose certain modules at the appropriate endpoints (like /oadrPoll and /EiRegister), and figure out what type of message is being sent. It will then call your handler with the contents of the message that are relevant for you to handle. This section provides an overview with examples for the different kinds of messages that you can expect and what should be returned.
on_register_report¶
The VEN informs you which reports it has available. If you want to periodically receive any of these reports, you should return a list of the r_ids that you want to receive.
Signature:
async def on_register_report(ven_id, resource_id, measurement, unit, scale,
min_sampling_interval, max_sampling_interval):
# If we want this report:
return (callback, requested_sampling_interval)
# or
return None
on_query_registration¶
A prospective VEN is requesting information about your VTN, like the versions and transports you support. You should not implement this handler and let OpenLEADR handle this response.
on_create_party_registration¶
The VEN tries to register with you. You will receive a registration_info dict that contains, among other things, a field ven_name which is how the VEN identifies itself. If the VEN is accepted, you return a ven_id, registration_id tuple. If not, return False:
async def on_create_party_registration(registration_info):
ven_name = registration_info['ven_name']
...
if ven_is_known:
return ven_id, registration_id
else
return None
During this step, the VEN probably does not have a venID yet. If they connected using a secure TLS connection, the registration_info dict will contain the fingerprint of the public key that was used for this connection (registration_info['fingerprint']). Your on_create_party_registration handler should check this fingerprint value against a value that you received offline, to be sure that the ven with this venName is the correct VEN.
on_cancel_party_registration¶
The VEN informs you that they are cancelling their registration and no longer wish to be contacted by you.
You should deregister the VEN internally, and return None.
Return: None
on_poll¶
You only need to implement this if you don’t want to use the automatic internal message queue. If you add this handler to the server, the internal message queue will be automatically disabled.
The VEN is requesting the next message that you have for it. You should return a tuple of message_type and message_payload as a dict. If there is no message for the VEN, you should return None.
Signature:
async def on_poll(ven_id):
...
return message_type, message_payload
If you implement your own on_poll handler, you should also include your own on_created_event handler that retrieves the opt status for a distributed event.
on_created_event¶
You only need to implement this if you don’t want to use the automatic internal message queue. Otherwise, you supply a per-event callback function when you add the event to the internal queue.
Signature:
async def on_created_event(ven_id, event_id, opt_status):
print("Ven {ven_id} returned {opt_status} for event {event_id}")
# return None
