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_hook(hook_name, handler)

You can add a hook at specific points in the request/reponse chain. Your choices are:

before_send_xml: you get the actual XML message just before it is sent over the wire after_receive_xml: you get the actual XML immediately after it is received over the wire before_parse_xml: you get the actual XML after its schema has been validated, but before parsing after_parse_xml: you get the message name and message payload after parsing the message

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_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 sync_events()

Used to perform an initial sync of events after the client connects

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: ProtocolError

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

Bases: 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: ProtocolError

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

Bases: ProtocolError

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

Bases: ProtocolError

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

Bases: ProtocolError

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

Bases: ProtocolError

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

Bases: ProtocolError

exception openleadr.errors.ProtocolError

Bases: Exception

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

Bases: 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: ProtocolError

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

Bases: 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

duration: 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: ActivePeriod = None

event_descriptor: EventDescriptor

event_signals: List[EventSignal]

response_required: str = 'always'

targets: List[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 = None

event_id: str

event_status: str

market_context: str

modification_date_time: 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[Interval]

measurement: Measurement = None

signal_id: str

signal_name: str

signal_type: str

targets: List[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

duration: 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: 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 = None

data_collection_mode: str = 'incremental'

dtstart: datetime = None

duration: timedelta = None

intervals: List[ReportInterval] = None

report_descriptions: List[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: Measurement = None

r_id: str

reading_type: str

report_data_source: Target

report_subject: Target

report_type: str

sampling_rate: SamplingRate

class openleadr.objects.ReportInterval(dtstart: datetime.datetime, report_payload: openleadr.objects.ReportPayload, duration: datetime.timedelta = None)

Bases: object

dtstart: datetime

duration: timedelta = None

report_payload: 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: 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: timedelta

report_back_duration: timedelta = None

report_interval: Interval = None

report_specifier_id: str

specifier_payloads: List[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: timedelta = None

min_period: timedelta = None

on_change: bool = False

class openleadr.objects.ServiceArea(feature_collection: openleadr.objects.FeatureCollection)

Bases: object

feature_collection: 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: 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: AggregatedPNode = None

end_device_asset: EndDeviceAsset = None

group_id: str = None

group_name: str = None

meter_asset: MeterAsset = None

p_node: PNode = None

party_id: str = None

resource_id: str = None

service_area: ServiceArea = None

service_delivery_point: ServiceDeliveryPoint = None

service_location: ServiceLocation = None

transport_interface: 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.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.