openleadr package

Submodules

openleadr.client module

class openleadr.client.OpenADRClient(ven_name, vtn_url, debug=False, cert=None, key=None, passphrase=None, vtn_fingerprint=None, show_fingerprint=True, ca_file=None, allow_jitter=True, ven_id=None, disable_signature=False, check_hostname=True)

Bases: object

Main client class. Most of these methods will be called automatically, but you can always choose to call them manually.

Initializes a new OpenADR Client (Virtual End Node)

Parameters
  • ven_name (str) – The name for this VEN

  • vtn_url (str) – The URL of the VTN (Server) to connect to

  • debug (bool) – Whether or not to print debugging messages

  • cert (str) – The path to a PEM-formatted Certificate file to use for signing messages.

  • key (str) – The path to a PEM-formatted Private Key file to use for signing messages.

  • passphrase (str) – The passphrase for the Private Key

  • vtn_fingerprint (str) – The fingerprint for the VTN’s certificate to verify incomnig messages

  • show_fingerprint (str) – Whether to print your own fingerprint on startup. Defaults to True.

  • ca_file (str) – The path to the PEM-formatted CA file for validating the VTN server’s certificate.

  • ven_id (str) – The ID for this VEN. If you leave this blank, a VEN_ID will be assigned by the VTN.

  • disable_signature (bool) – Whether or not to sign outgoing messages using a public-private key pair in PEM format.

add_handler(handler, callback)

Add a callback for the given situation

add_report(callback, resource_id, measurement=None, data_collection_mode='incremental', report_specifier_id=None, r_id=None, report_name='TELEMETRY_USAGE', reading_type='Direct Read', report_type='reading', report_duration=None, report_dtstart=None, sampling_rate=None, data_source=None, scale='none', unit=None, power_ac=True, power_hertz=50, power_voltage=230, market_context=None, end_device_asset_mrid=None, report_data_source=None)

Add a new reporting capability to the client.

Parameters
  • callback (callable) – A callback or coroutine that will fetch the value for a specific report. This callback will be passed the report_id and the r_id of the requested value.

  • resource_id (str) – A specific name for this resource within this report.

  • measurement (str) – The quantity that is being measured (openleadr.enums.MEASUREMENTS). Optional for TELEMETRY_STATUS reports.

  • data_collection_mode (str) – Whether you want the data to be collected incrementally or at once. If the VTN requests the sampling interval to be higher than the reporting interval, this setting determines if the callback should be called at the sampling rate (with no args, assuming it returns the current value), or at the reporting interval (with date_from and date_to as keyword arguments). Choose ‘incremental’ for the former case, or ‘full’ for the latter case.

  • report_specifier_id (str) – A unique identifier for this report. Leave this blank for a random generated id, or fill it in if your VTN depends on this being a known value, or if it needs to be constant between restarts of the client.

  • r_id (str) – A unique identifier for a datapoint in a report. The same remarks apply as for the report_specifier_id.

  • report_name (str) – An OpenADR name for this report (one of openleadr.enums.REPORT_NAME)

  • reading_type (str) – An OpenADR reading type (found in openleadr.enums.READING_TYPE)

  • report_type (str) – An OpenADR report type (found in openleadr.enums.REPORT_TYPE)

  • report_duration (datetime.timedelta) – The time span that can be provided in this report.

  • report_dtstart (datetime.datetime) – The earliest available data for this report (defaults to now).

  • sampling_rate (datetime.timedelta) – The sampling rate for the measurement.

  • unit (str) – The unit for this measurement.

  • power_ac (boolean) – Whether the power is AC (True) or DC (False). Only required when supplying a power-related measurement.

  • power_hertz (int) – Grid frequency of the power. Only required when supplying a power-related measurement.

  • power_voltage (int) – Voltage of the power. Only required when supplying a power-related measurement.

  • market_context (str) – The Market Context that this report belongs to.

  • end_device_asset_mrid (str) – the Meter ID for the end device that is measured by this report.

  • report_data_source – A (list of) target(s) that this report is related to.

async cancel_party_registration()
async cancel_report(payload)

Cancel this report.

async create_party_registration(http_pull_model=True, xml_signature=False, report_only=False, profile_name='2.0b', transport_name='simpleHttp', transport_address=None, ven_id=None)

Take the neccessary steps to register this client with the server.

Parameters
  • http_pull_model (bool) – Whether to use the ‘pull’ model for HTTP.

  • xml_signature (bool) – Whether to sign each XML message.

  • report_only (bool) – Whether or not this is a reporting-only client which does not deal with Events.

  • profile_name (str) – Which OpenADR profile to use.

  • transport_name (str) – The transport name to use. Either ‘simpleHttp’ or ‘xmpp’.

  • transport_address (str) – Which public-facing address the server should use to communicate.

async create_report(report_request)

Add the requested reports to the reporting mechanism. This is called when the VTN requests reports from us.

Parameters

dict (report_request) – The oadrReportRequest dict from the VTN.

async create_single_report(report_request)

Create a single report in response to a request from the VTN.

async created_event(request_id, event_id, opt_type, modification_number=0)

Inform the VTN that we created an event.

async on_cancel_party_registration(message)
async on_event(event)

Placeholder for the on_event handler.

async on_register_report(report)

Placeholder for the on_register_report handler.

async on_update_event(event)

Placeholder for the on_update_event handler.

async poll()

Request the next available message from the Server. This coroutine is called automatically.

async query_registration()

Request information about the VTN.

async register_reports(reports)

Tell the VTN about our reports. The VTN miht respond with an oadrCreateReport message that tells us which reports are to be sent.

async request_event(reply_limit=None)

Request the next Event from the VTN, if it has any.

async run()

Run the client in full-auto mode.

async send_response(service, response_code=200, response_description='OK', request_id=None)

Send an empty oadrResponse, for instance after receiving oadrRequestReregistration.

async stop()

Cleanly stops the client. Run this coroutine before closing your event loop.

async update_report(report_request_id)

Call the previously registered report callback and send the result as a message to the VTN.

openleadr.enums module

A collection of useful enumerations that you can use to construct or interpret OpenADR messages. Can also be useful during testing.

class openleadr.enums.EVENT_STATUS

Bases: object

ACTIVE = 'active'
CANCELLED = 'cancelled'
COMPLETED = 'completed'
FAR = 'far'
NEAR = 'near'
NONE = 'none'
class openleadr.enums.Enum

Bases: type

property members
property values
class openleadr.enums.MEASUREMENTS

Bases: object

ACTIVE_ENERGY = Measurement(name='energyReal', description='RealEnergy', unit='Wh', scale='none', power_attributes=None, pulse_factor=None, ns='power')
ACTIVE_POWER = Measurement(name='powerReal', description='RealPower', unit='W', scale='none', power_attributes=PowerAttributes(hertz=50, voltage=230, ac=True), pulse_factor=None, ns='power')
APPARENT_ENERGY = Measurement(name='energyApparent', description='ApparentEnergy', unit='VAh', scale='none', power_attributes=None, pulse_factor=None, ns='power')
APPARENT_POWER = Measurement(name='powerApparent', description='ApparentPower', unit='VA', scale='none', power_attributes=PowerAttributes(hertz=50, voltage=230, ac=True), pulse_factor=None, ns='power')
CURRENCY = Measurement(name='currency', description='currency', unit='AFN', scale='none', power_attributes=None, pulse_factor=None, ns='oadr')
CURRENCY_PER_KW = Measurement(name='customUnit', description='currencyPerKW', unit='AFN', scale='none', power_attributes=None, pulse_factor=None, ns='oadr')
CURRENCY_PER_KWH = Measurement(name='currencyPerKWh', description='currencyPerKWh', unit='AFN', scale='none', power_attributes=None, pulse_factor=None, ns='oadr')
CURRENCY_PER_THM = Measurement(name='currencyPerThm', description='currency', unit='AFN', scale='none', power_attributes=None, pulse_factor=None, ns='oadr')
CURRENT = Measurement(name='current', description='Current', unit='A', scale='none', power_attributes=None, pulse_factor=None, ns='oadr')
ENERGY_APPARENT = Measurement(name='energyApparent', description='ApparentEnergy', unit='VAh', scale='none', power_attributes=None, pulse_factor=None, ns='power')
ENERGY_REACTIVE = Measurement(name='energyReactive', description='ReactiveEnergy', unit='VARh', scale='none', power_attributes=None, pulse_factor=None, ns='power')
ENERGY_REAL = Measurement(name='energyReal', description='RealEnergy', unit='Wh', scale='none', power_attributes=None, pulse_factor=None, ns='power')
FREQUENCY = Measurement(name='frequency', description='Frequency', unit='Hz', scale='none', power_attributes=None, pulse_factor=None, ns='oadr')
POWER_APPARENT = Measurement(name='powerApparent', description='ApparentPower', unit='VA', scale='none', power_attributes=PowerAttributes(hertz=50, voltage=230, ac=True), pulse_factor=None, ns='power')
POWER_REACTIVE = Measurement(name='powerReactive', description='ReactivePower', unit='VAR', scale='none', power_attributes=PowerAttributes(hertz=50, voltage=230, ac=True), pulse_factor=None, ns='power')
POWER_REAL = Measurement(name='powerReal', description='RealPower', unit='W', scale='none', power_attributes=PowerAttributes(hertz=50, voltage=230, ac=True), pulse_factor=None, ns='power')
PULSE_COUNT = Measurement(name='pulseCount', description='pulse count', unit='count', scale=None, power_attributes=None, pulse_factor=1000, ns='oadr')
REACTIVE_ENERGY = Measurement(name='energyReactive', description='ReactiveEnergy', unit='VARh', scale='none', power_attributes=None, pulse_factor=None, ns='power')
REACTIVE_POWER = Measurement(name='powerReactive', description='ReactivePower', unit='VAR', scale='none', power_attributes=PowerAttributes(hertz=50, voltage=230, ac=True), pulse_factor=None, ns='power')
REAL_ENERGY = Measurement(name='energyReal', description='RealEnergy', unit='Wh', scale='none', power_attributes=None, pulse_factor=None, ns='power')
REAL_POWER = Measurement(name='powerReal', description='RealPower', unit='W', scale='none', power_attributes=PowerAttributes(hertz=50, voltage=230, ac=True), pulse_factor=None, ns='power')
TEMPERATURE = Measurement(name='temperature', description='temperature', unit='celsius', scale='none', power_attributes=None, pulse_factor=None, ns='oadr')
THERM = Measurement(name='Therm', description='Therm', unit='thm', scale='none', power_attributes=None, pulse_factor=None, ns='oadr')
VOLTAGE = Measurement(name='voltage', description='Voltage', unit='V', scale='none', power_attributes=None, pulse_factor=None, ns='power')
class openleadr.enums.OPT

Bases: object

OPT_IN = 'optIn'
OPT_OUT = 'optOut'
class openleadr.enums.OPT_REASON

Bases: object

ECONOMIC = 'economic'
EMERGENCY = 'emergency'
MUST_RUN = 'mustRun'
NOT_PARTICIPATING = 'notParticipating'
OUTAGE_RUN_STATUS = 'outageRunStatus'
OVERRIDE_STATUS = 'overrideStatus'
PARTICIPATING = 'participating'
X_SCHEDULE = 'x-schedule'
class openleadr.enums.READING_TYPE

Bases: object

ALLOCATED = 'Allocated'
CONTRACT = 'Contract'
DERIVED = 'Derived'
DIRECT_READ = 'Direct Read'
ESTIMATED = 'Estimated'
HYBRID = 'Hybrid'
MEAN = 'Mean'
NET = 'Net'
PEAK = 'Peak'
PROJECTED = 'Projected'
SUMMED = 'Summed'
X_NOT_APPLICABLE = 'x-notApplicable'
X_RMS = 'x-RMS'
class openleadr.enums.REPORT_NAME

Bases: object

HISTORY_GREENBUTTON = 'HISTORY_GREENBUTTON'
HISTORY_USAGE = 'HISTORY_USAGE'
METADATA_HISTORY_GREENBUTTON = 'METADATA_HISTORY_GREENBUTTON'
METADATA_HISTORY_USAGE = 'METADATA_HISTORY_USAGE'
METADATA_TELEMETRY_STATUS = 'METADATA_TELEMETRY_STATUS'
METADATA_TELEMETRY_USAGE = 'METADATA_TELEMETRY_USAGE'
TELEMETRY_STATUS = 'TELEMETRY_STATUS'
TELEMETRY_USAGE = 'TELEMETRY_USAGE'
class openleadr.enums.REPORT_TYPE

Bases: object

AVAILABLE_ENERGY_STORAGE = 'availableEnergyStorage'
AVG_DEMAND = 'avgDemand'
AVG_USAGE = 'avgUsage'
BASELINE = 'baseline'
DELTA_DEMAND = 'deltaDemand'
DELTA_SET_POINT = 'deltaSetPoint'
DELTA_USAGE = 'deltaUsage'
DEMAND = 'demand'
DEVIATION = 'deviation'
DOWN_REGULATION_CAPACITY_AVAILABLE = 'downRegulationCapacityAvailable'
LEVEL = 'level'
OPERATING_STATE = 'operatingState'
PERCENT_DEMAND = 'percentDemand'
PERCENT_USAGE = 'percentUsage'
POWER_FACTOR = 'powerFactor'
PRICE = 'price'
READING = 'reading'
REGULATION_SETPOINT = 'regulationSetpoint'
SET_POINT = 'setPoint'
STORED_ENERGY = 'storedEnergy'
TARGET_ENERGY_STORAGE = 'targetEnergyStorage'
UP_REGULATION_CAPACITY_AVAILABLE = 'upRegulationCapacityAvailable'
USAGE = 'usage'
X_RESOURCE_STATUS = 'x-resourceStatus'
class openleadr.enums.SECURITY_LEVEL

Bases: object

HIGH = 'HIGH'
STANDARD = 'STANDARD'
class openleadr.enums.SIGNAL_NAME

Bases: object

BID_ENERGY = 'BID_ENERGY'
BID_LOAD = 'BID_LOAD'
BID_PRICE = 'BID_PRICE'
CHARGE_STATE = 'CHARGE_STATE'
DEMAND_CHARGE = 'DEMAND_CHARGE'
ELECTRICITY_PRICE = 'ELECTRICITY_PRICE'
ENERGY_PRICE = 'ENERGY_PRICE'
LOAD_CONTROL = 'LOAD_CONTROL'
LOAD_DISPATCH = 'LOAD_DISPATCH'
SIMPLE = 'SIMPLE'
simple = 'simple'
class openleadr.enums.SIGNAL_TARGET_MRID

Bases: object

BASEBOARD_HEATER = 'Baseboard_Heater'
ELECTRIC_VEHICLE = 'Electric_Vehicle'
ENERGY_MANAGEMENT_SYSTEM = 'Energy_Management_System'
EVSE = 'EVSE'
EXTERIOR_LIGHTING = 'Exterior_Lighting'
GENERATION_SYSTEMS = 'Generation_Systems'
HOT_TUB = 'Hot_tub'
INTERIOR_LIGHTING = 'Interior_Lighting'
IRRIGATION_PUMP = 'Irrigation_Pump'
LOAD_CONTROL_SWITCH = 'Load_Control_Switch'
MANAGED_COMMERCIAL_AND_INDUSTRIAL_LOADS = 'Managed_Commercial_and_Industrial_Loads'
POOL_PUMP = 'Pool_Pump'
RESU = 'RESU'
SAUNA = 'Sauna'
SIMPLE_RESIDENTIAL_ON_OFF_LOADS = 'Simple_Residential_On_Off_Loads'
SMART_APPLIANCE = 'Smart_Appliance'
SMART_ENERGY_MODULE = 'Smart_Energy_Module'
SMART_INVERTER = 'Smart_Inverter'
STORAGE = 'Storage'
STRIP_HEATER = 'Strip_Heater'
THERMOSTAT = 'Thermostat'
WATER_HEATER = 'Water_Heater'
class openleadr.enums.SIGNAL_TYPE

Bases: object

DELTA = 'delta'
LEVEL = 'level'
MULTIPLIER = 'multiplier'
PRICE = 'price'
PRICE_MULTIPLIER = 'priceMultiplier'
PRICE_RELATIVE = 'priceRelative'
SETPOINT = 'setpoint'
X_LOAD_CONTROL_CAPACITY = 'x-loadControlCapacity'
X_LOAD_CONTROL_LEVEL_OFFSET = 'x-loadControlLevelOffset'
X_LOAD_CONTROL_PERCENT_OFFSET = 'x-loadControlPercentOffset'
X_LOAD_CONTROL_SETPOINT = 'x-loadControlSetpoint'
class openleadr.enums.SI_SCALE_CODE

Bases: object

G = 'G'
M = 'M'
T = 'T'
c = 'c'
d = 'd'
k = 'k'
m = 'm'
micro = 'micro'
n = 'n'
none = 'none'
p = 'p'
class openleadr.enums.STATUS_CODES

Bases: object

COMPLIANCE_ERROR = 459
DEPLOYMENT_ERROR_OR_OTHER_ERROR = 469
INVALID_DATA = 454
INVALID_ID = 452
NOT_ALLOWED = 451
NOT_RECOGNIZED = 453
NOT_REGISTERED_OR_AUTHORIZED = 463
OUT_OF_SEQUENCE = 450
REPORT_NOT_SUPPORTED = 461
SIGNAL_NOT_SUPPORTED = 460
TARGET_MISMATCH = 462

openleadr.errors module

exception openleadr.errors.ComplianceError(description='COMPLIANCE ERROR')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.DeploymentError(description='DEPLOYMENT ERROR OR OTHER ERROR')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.FingerprintMismatch

Bases: Exception

exception openleadr.errors.HTTPError(status=500, description=None)

Bases: Exception

exception openleadr.errors.InvalidDataError(description='INVALID DATA')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.InvalidIdError(description='INVALID ID')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.NotAllowedError(description='NOT ALLOWED')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.NotRecognizedError(description='NOT RECOGNIZED')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.NotRegisteredOrAuthorizedError(description='NOT REGISTERED OR AUTHORIZED')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.OutOfSequenceError(description='OUT OF SEQUENCE')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.ProtocolError

Bases: Exception

exception openleadr.errors.ReportNotSupportedError(description='REPORT NOT SUPPORTED')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.RequestReregistration(ven_id=None)

Bases: Exception

exception openleadr.errors.SendEmptyHTTPResponse

Bases: Exception

exception openleadr.errors.SignalNotSupportedError(description='SIGNAL NOT SUPPORTED')

Bases: openleadr.errors.ProtocolError

exception openleadr.errors.TargetMismatchError(description='TARGET MISMATCH')

Bases: openleadr.errors.ProtocolError

openleadr.fingerprint module

openleadr.fingerprint.show_fingerprint()

openleadr.hooks module

openleadr.hooks.call(hook_point, *args, **kwargs)
openleadr.hooks.register(hook_point, callback)

Call a hook

openleadr.messaging module

async openleadr.messaging.authenticate_message(request, message_tree, message_payload, fingerprint_lookup=None, ven_lookup=None)
openleadr.messaging.create_message(message_type, cert=None, key=None, passphrase=None, disable_signature=False, **message_payload)

Create and optionally sign an OpenADR message. Returns an XML string.

openleadr.messaging.parse_message(data)

Parse a message and distill its usable parts. Returns a message type and payload. :param data str: The XML string that is received

Returns a message type (str) and a message payload (dict)

openleadr.messaging.validate_xml_schema(content)

Validates the XML tree against the schema. Return the XML tree.

openleadr.messaging.validate_xml_signature(xml_tree, cert_fingerprint=None)

Validate the XMLDSIG signature and the ReplayProtect element.

openleadr.messaging.validate_xml_signature_none(xml_tree)

openleadr.objects module

class openleadr.objects.ActivePeriod(dtstart: datetime.datetime, duration: datetime.timedelta, tolerance: dict = None, notification_period: dict = None, ramp_up_period: dict = None, recovery_period: dict = None)

Bases: object

dtstart: datetime.datetime
duration: datetime.timedelta
notification_period: dict = None
ramp_up_period: dict = None
recovery_period: dict = None
tolerance: dict = None
class openleadr.objects.AggregatedPNode(node: str)

Bases: object

node: str
class openleadr.objects.EndDeviceAsset(mrid: str)

Bases: object

mrid: str
class openleadr.objects.Event(event_descriptor: openleadr.objects.EventDescriptor, event_signals: List[openleadr.objects.EventSignal], targets: List[openleadr.objects.Target] = None, targets_by_type: Dict = None, active_period: openleadr.objects.ActivePeriod = None, response_required: str = 'always')

Bases: object

active_period: openleadr.objects.ActivePeriod = None
event_descriptor: openleadr.objects.EventDescriptor
event_signals: List[openleadr.objects.EventSignal]
response_required: str = 'always'
targets: List[openleadr.objects.Target] = None
targets_by_type: Dict = None
class openleadr.objects.EventDescriptor(event_id: str, modification_number: int, market_context: str, event_status: str, created_date_time: datetime.datetime = None, modification_date_time: datetime.datetime = None, priority: int = 0, test_event: bool = False, vtn_comment: str = None)

Bases: object

created_date_time: datetime.datetime = None
event_id: str
event_status: str
market_context: str
modification_date_time: datetime.datetime = None
modification_number: int
priority: int = 0
test_event: bool = False
vtn_comment: str = None
class openleadr.objects.EventSignal(intervals: List[openleadr.objects.Interval], signal_name: str, signal_type: str, signal_id: str, current_value: float = None, targets: List[openleadr.objects.Target] = None, targets_by_type: Dict = None, measurement: openleadr.objects.Measurement = None)

Bases: object

current_value: float = None
intervals: List[openleadr.objects.Interval]
measurement: openleadr.objects.Measurement = None
signal_id: str
signal_name: str
signal_type: str
targets: List[openleadr.objects.Target] = None
targets_by_type: Dict = None
class openleadr.objects.FeatureCollection(id: str, location: dict)

Bases: object

id: str
location: dict
class openleadr.objects.Interval(dtstart: datetime.datetime, duration: datetime.timedelta, signal_payload: float, uid: int = None)

Bases: object

dtstart: datetime.datetime
duration: datetime.timedelta
signal_payload: float
uid: int = None
class openleadr.objects.Measurement(name: str, description: str, unit: str, acceptable_units: List[str] = <factory>, scale: str = None, power_attributes: openleadr.objects.PowerAttributes = None, pulse_factor: int = None, ns: str = 'power')

Bases: object

acceptable_units: List[str]
description: str
name: str
ns: str = 'power'
power_attributes: openleadr.objects.PowerAttributes = None
pulse_factor: int = None
scale: str = None
unit: str
class openleadr.objects.MeterAsset(mrid: str)

Bases: object

mrid: str
class openleadr.objects.PNode(node: str)

Bases: object

node: str
class openleadr.objects.PowerAttributes(hertz: int = 50, voltage: int = 230, ac: bool = True)

Bases: object

ac: bool = True
hertz: int = 50
voltage: int = 230
class openleadr.objects.Report(report_specifier_id: str, report_name: str, report_request_id: str = None, report_descriptions: List[openleadr.objects.ReportDescription] = None, created_date_time: datetime.datetime = None, dtstart: datetime.datetime = None, duration: datetime.timedelta = None, intervals: List[openleadr.objects.ReportInterval] = None, data_collection_mode: str = 'incremental')

Bases: object

created_date_time: datetime.datetime = None
data_collection_mode: str = 'incremental'
dtstart: datetime.datetime = None
duration: datetime.timedelta = None
intervals: List[openleadr.objects.ReportInterval] = None
report_descriptions: List[openleadr.objects.ReportDescription] = None
report_name: str
report_request_id: str = None
report_specifier_id: str
class openleadr.objects.ReportDescription(r_id: str, market_context: str, reading_type: str, report_subject: openleadr.objects.Target, report_data_source: openleadr.objects.Target, report_type: str, sampling_rate: openleadr.objects.SamplingRate, measurement: openleadr.objects.Measurement = None)

Bases: object

market_context: str
measurement: openleadr.objects.Measurement = None
r_id: str
reading_type: str
report_data_source: openleadr.objects.Target
report_subject: openleadr.objects.Target
report_type: str
sampling_rate: openleadr.objects.SamplingRate
class openleadr.objects.ReportInterval(dtstart: datetime.datetime, report_payload: openleadr.objects.ReportPayload, duration: datetime.timedelta = None)

Bases: object

dtstart: datetime.datetime
duration: datetime.timedelta = None
report_payload: openleadr.objects.ReportPayload
class openleadr.objects.ReportPayload(r_id: str, value: float, confidence: int = None, accuracy: int = None)

Bases: object

accuracy: int = None
confidence: int = None
r_id: str
value: float
class openleadr.objects.ReportRequest(report_request_id: str, report_specifier: openleadr.objects.ReportSpecifier)

Bases: object

report_request_id: str
report_specifier: openleadr.objects.ReportSpecifier
class openleadr.objects.ReportSpecifier(report_specifier_id: str, granularity: datetime.timedelta, specifier_payloads: List[openleadr.objects.SpecifierPayload], report_interval: openleadr.objects.Interval = None, report_back_duration: datetime.timedelta = None)

Bases: object

granularity: datetime.timedelta
report_back_duration: datetime.timedelta = None
report_interval: openleadr.objects.Interval = None
report_specifier_id: str
specifier_payloads: List[openleadr.objects.SpecifierPayload]
class openleadr.objects.Response(response_code: int, response_description: str, request_id: str)

Bases: object

request_id: str
response_code: int
response_description: str
class openleadr.objects.SamplingRate(min_period: datetime.timedelta = None, max_period: datetime.timedelta = None, on_change: bool = False)

Bases: object

max_period: datetime.timedelta = None
min_period: datetime.timedelta = None
on_change: bool = False
class openleadr.objects.ServiceArea(feature_collection: openleadr.objects.FeatureCollection)

Bases: object

feature_collection: openleadr.objects.FeatureCollection
class openleadr.objects.ServiceDeliveryPoint(node: str)

Bases: object

node: str
class openleadr.objects.ServiceLocation(node: str)

Bases: object

node: str
class openleadr.objects.SpecifierPayload(r_id: str, reading_type: str, measurement: openleadr.objects.Measurement = None)

Bases: object

measurement: openleadr.objects.Measurement = None
r_id: str
reading_type: str
class openleadr.objects.Target(aggregated_p_node: openleadr.objects.AggregatedPNode = None, end_device_asset: openleadr.objects.EndDeviceAsset = None, meter_asset: openleadr.objects.MeterAsset = None, p_node: openleadr.objects.PNode = None, service_area: openleadr.objects.ServiceArea = None, service_delivery_point: openleadr.objects.ServiceDeliveryPoint = None, service_location: openleadr.objects.ServiceLocation = None, transport_interface: openleadr.objects.TransportInterface = None, group_id: str = None, group_name: str = None, resource_id: str = None, ven_id: str = None, party_id: str = None)

Bases: object

aggregated_p_node: openleadr.objects.AggregatedPNode = None
end_device_asset: openleadr.objects.EndDeviceAsset = None
group_id: str = None
group_name: str = None
meter_asset: openleadr.objects.MeterAsset = None
p_node: openleadr.objects.PNode = None
party_id: str = None
resource_id: str = None
service_area: openleadr.objects.ServiceArea = None
service_delivery_point: openleadr.objects.ServiceDeliveryPoint = None
service_location: openleadr.objects.ServiceLocation = None
transport_interface: openleadr.objects.TransportInterface = None
ven_id: str = None
class openleadr.objects.TransportInterface(point_of_receipt: str, point_of_delivery: str)

Bases: object

point_of_delivery: str
point_of_receipt: str

openleadr.preflight module

openleadr.preflight.preflight_message(message_type, message_payload)

Tests message contents before sending them. It will correct benign errors and warn you about them. Uncorrectable errors will raise an Exception. It changes the message_payload dict in-place.

Parameters
  • string (message_type) – The type of message you are sending

  • dict (message_payload) – The contents of the message

openleadr.provisioning module

openleadr.provisioning.generate_certificates(ven_id, passphrase, root_ca, **kwargs)

Generate a PEM keypair for a VEN, signed by the given root ca.

openleadr.provisioning.sign_csr(csr_str, root_ca)

Sign the CSR with our own root ca

openleadr.server module

class openleadr.server.OpenADRServer(vtn_id, cert=None, key=None, passphrase=None, fingerprint_lookup=None, show_fingerprint=True, http_port=8080, http_host='127.0.0.1', http_cert=None, http_key=None, http_key_passphrase=None, http_path_prefix='/OpenADR2/Simple/2.0b', requested_poll_freq=datetime.timedelta(seconds=10), http_ca_file=None, ven_lookup=None)

Bases: object

Create a new OpenADR VTN (Server).

Parameters
  • vtn_id (str) – An identifier string for this VTN. This is how you identify yourself to the VENs that talk to you.

  • cert (str) – Path to the PEM-formatted certificate file that is used to sign outgoing messages

  • key (str) – Path to the PEM-formatted private key file that is used to sign outgoing messages

  • passphrase (str) – The passphrase used to decrypt the private key file

  • fingerprint_lookup (callable) – A callable that receives a ven_id and should return the registered fingerprint for that VEN. You should receive these fingerprints outside of OpenADR and configure them manually.

  • show_fingerprint (bool) – Whether to print the fingerprint to your stdout on startup. Defaults to True.

  • http_port (int) – The port that the web server is exposed on (default: 8080)

  • http_host (str) – The host or IP address to bind the server to (default: 127.0.0.1).

  • http_cert (str) – The path to the PEM certificate for securing HTTP traffic.

  • http_key (str) – The path to the PEM private key for securing HTTP traffic.

  • http_ca_file (str) – The path to the CA-file that client certificates are checked against.

  • http_key_passphrase (str) – The passphrase for the HTTP private key.

  • ven_lookup – A callback that takes a ven_id and returns a dict containing the ven_id, ven_name, fingerprint and registration_id.

add_event(ven_id, signal_name, signal_type, intervals, callback=None, event_id=None, targets=None, targets_by_type=None, target=None, response_required='always', market_context='oadr://unknown.context', notification_period=None, ramp_up_period=None, recovery_period=None, signal_target_mrid=None)

Convenience method to add an event with a single signal.

Parameters
  • ven_id (str) – The ven_id to whom this event must be delivered.

  • signal_name (str) – The OpenADR name of the signal; one of openleadr.objects.SIGNAL_NAME

  • signal_type (str) – The OpenADR type of the signal; one of openleadr.objects.SIGNAL_TYPE

  • intervals (str) – A list of intervals with a dtstart, duration and payload member.

  • callback (str) – A callback function for when your event has been accepted (optIn) or refused (optOut).

  • targets (list) – A list of Targets that this Event applies to.

  • target – A single target for this event.

  • targets_by_type (dict) – A dict of targets, grouped by type.

  • market_context (str) – A URI for the DR program that this event belongs to.

  • notification_period (timedelta) – The Notification period for the Event’s Active Period.

  • ramp_up_period (timedelta) – The Ramp Up period for the Event’s Active Period.

  • recovery_period (timedelta) – The Recovery period for the Event’s Active Period.

If you don’t provide a target using any of the three arguments, the target will be set to the given ven_id.

add_handler(name, func)

Add a handler to the OpenADRServer.

Parameters
  • name (str) – The name for this handler. Should be one of: on_created_event, on_request_event, on_register_report, on_create_report, on_created_report, on_request_report, on_update_report, on_poll, on_query_registration, on_create_party_registration, on_cancel_party_registration.

  • func (callable) – A function or coroutine that handles this type of occurrence. It receives the message, and should return the contents of a response.

add_raw_event(ven_id, event, callback=None)

Add a new event to the queue for a specific VEN. :param str ven_id: The ven_id to which this event should be distributed. :param dict event: The event (as a dict or as a objects.Event instance)

that contains the event details.

Parameters

callback (callable) – A callback that will receive the opt status for this event. This callback receives ven_id, event_id, opt_type as its arguments.

cancel_event(ven_id, event_id)

Mark the indicated event as cancelled.

property event_callbacks
property events
property events_updated
property registered_reports
async run()

Starts the server in an already-running asyncio loop.

async run_async()
async stop()

Stop the server in a graceful manner.

openleadr.utils module

async openleadr.utils.await_if_required(result)
openleadr.utils.booleanformat(value)

Format a boolean value

openleadr.utils.certificate_fingerprint(certificate_str)

Calculate the fingerprint for the given certificate, as defined by OpenADR.

openleadr.utils.certificate_fingerprint_from_der(der_bytes)
openleadr.utils.cron_config(interval, randomize_seconds=False)

Returns a dict with cron settings for the given interval

openleadr.utils.datetimeformat(value, format='%Y-%m-%dT%H:%M:%S.%fZ')

Format a given datetime as a UTC ISO3339 string.

openleadr.utils.determine_event_status(active_period)
openleadr.utils.ensure_bytes(obj)

Converts a utf-8 str object to bytes.

openleadr.utils.ensure_str(obj)

Converts bytes to a utf-8 string.

openleadr.utils.extract_pem_cert(tree)

Extract a given X509 certificate inside an XML tree and return the standard form of a PEM-encoded certificate.

Parameters

lxml.etree (tree) – The tree that contains the X509 element. This is usually the KeyInfo element from the XMLDsig Signature part of the message.

openleadr.utils.find_by(dict_or_list, key, value, *args)

Find a dict inside a dict or list by key, value properties. You can search for a nesting by separating the levels with a period (.).

openleadr.utils.flatten_xml(message)

Flatten the entire XML structure.

async openleadr.utils.gather_if_required(results)
openleadr.utils.generate_id(*args, **kwargs)

Generate a string that can be used as an identifier in OpenADR messages.

openleadr.utils.get_active_period_from_intervals(intervals, as_dict=True)
openleadr.utils.get_cert_fingerprint_from_request(request)
openleadr.utils.getmember(obj, member, missing='_RAISE_')

Get a member from a dict or dataclass. Nesting is possible.

openleadr.utils.group_by(list_, key, pop_key=False)

Return a dict that groups values

openleadr.utils.group_targets_by_type(list_of_targets)
openleadr.utils.hasmember(obj, member)

Check if a dict or dataclass has the given member

openleadr.utils.increment_event_modification_number(event)

Increments the modification number of the event by 1 and returns the new modification number.

openleadr.utils.normalize_dict(ordered_dict)

Main conversion function for the output of xmltodict to the OpenLEADR representation of OpenADR contents.

Parameters

dict (ordered_dict) – The OrderedDict, dict or dataclass that you wish to convert.

openleadr.utils.order_events(events, limit=None, offset=None)

Order the events according to the OpenADR rules: - active events before inactive events - high priority before low priority - earlier before later

openleadr.utils.parse_boolean(value)
openleadr.utils.parse_datetime(value)

Parse an ISO8601 datetime into a datetime.datetime object.

openleadr.utils.parse_duration(value)

Parse a RFC5545 duration.

openleadr.utils.pop_by(list_, key, value, *args)

Pop the first item that satisfies the search params from the given list.

openleadr.utils.setmember(obj, member, value)

Set a member of a dict of dataclass

openleadr.utils.timedeltaformat(value)

Format a timedelta to a RFC5545 Duration.

openleadr.utils.ungroup_targets_by_type(targets_by_type)
openleadr.utils.validate_report_measurement_dict(measurement)
openleadr.utils.validate_report_request_tuples(list_of_report_requests, mode='full')

Module contents

openleadr.enable_default_logging(level=20)

Turn on logging to stdout. :param level integer: The logging level you wish to use.

Defaults to logging.INFO.