Homec4science
Diffusion EMS for Smart-Building (master) Browse

R3852/rtevents_interface/rtevents_interf.py
ebd58783eab5master

Last Change
Raw File

rtevents_interf.py
View Options

__author__ = 'Olivier Van Cutsem'
import time
from multiprocessing import Queue
import Queue
from threading import Thread
from ems_config import *
import zmq.green as zmq
from bmsinterface.bms_interface_config import DB_KEY_EMS_SENSORS_CONVERSION
from bmsinterface.bms_restful_client import *
import json
class RTEventInterface(Thread):
"""
Description and how to use it
"""
def __init__(self, out_q, in_q):
"""
Initialize BMSInterface object.
Launch a thread to listen to BMS bubble messages through a websocket
Initialize the dictionary to link sensors and loads
Filter the incoming messages
"""
Thread.__init__(self)
self.out_stream = out_q
self.in_stream = in_q
# ZMQ sockets?
if RT_EVENT_PROTOCOL == RT_EVENT_PROTOCOL_ZMQ:
zmq_context = zmq.Context()
self.rtevent_sub = zmq_context.socket(zmq.SUB)
self.rtevent_sub.setsockopt(zmq.SUBSCRIBE, b'')
self.rtevent_pub = zmq_context.socket(zmq.PUB)
self.rtevent_pub.setsockopt(zmq.LINGER, -1)
self.socket_poller = zmq.Poller()
elif RT_EVENT_PROTOCOL == RT_EVENT_PROTOCOL_WEBSOCKET:
#TODO
pass
# Simulation mode? -> direct connexion with the vEngine
if EMS_MODE == EMS_MODE_VE_SIMULATION:
self.ve_to_bms_mapping = dict()
self.bms_to_ve_mapping = dict()
def run(self):
""" ___ """
# Initialize the interface
self.configure_interface()
# Main loop
while True:
# Messages coming from the BMS interface
msg_in = self.get_in_message()
if msg_in is not None:
processed_msg = self.process_in_event(msg_in)
encoded_msg = processed_msg # TODO: specify the type of message (simu, sensor event, other?)
for msg in encoded_msg:
self.out_stream.put(msg)
# Messages coming from the EMS interface
try:
msg_out = self.in_stream.get(False)
except Queue.Empty:
msg_out = None
if msg_out is not None:
self.send_downstream_event(msg_out)
### @ Initialization
def configure_interface(self):
"""
Configure the interface in use
:return: /
"""
if RT_EVENT_PROTOCOL == RT_EVENT_PROTOCOL_ZMQ:
self.rtevent_sub.connect('tcp://' + ZMQ_BMS_IP + ':' + ZMQ_BMS_PORT_PUB)
self.rtevent_pub.connect('tcp://' + ZMQ_BMS_IP + ':' + ZMQ_BMS_PORT_SUB)
self.socket_poller.register(self.rtevent_sub, zmq.POLLIN)
time.sleep(SOCKET_SETTLE_TIME) # Necessary for the sockets to settle.
print("ZMQ sockets created")
elif RT_EVENT_PROTOCOL == RT_EVENT_PROTOCOL_WEBSOCKET:
#TODO
pass
# In the EMS in connected directly to the vEngine, a table has to be instantiated in order to map the
# vE_ID to their corresponding sensor/actuator BMS ID
#todo: encode the keys of the BMS
if EMS_MODE == EMS_MODE_VE_SIMULATION:
list_of_vEntities = get_simulation_entities()
for ve in list_of_vEntities:
ref = ve['pk'] # BMS id of the vEntity AND ref as well !
data_ve = ve['fields']
list_of_sensors = None
if data_ve['sensor'] is not None: # comfort sensor?
data_sensor = data_ve['sensor']
#todo: normally ve['sensor'] should be a list !
list_of_sensors = {DB_KEY_EMS_SENSORS_CONVERSION[data_sensor['fields']['measure']]: data_sensor['pk']}
elif data_ve['actuator'] is not None: # comfort actuator?
data_actuator = data_ve['actuator']
#todo: normally ve['actuator'] should be a list !
# data_actuator['fields']['action'] should be in ACTUATOR_API_TYPES
list_of_sensors = {data_actuator['fields']['action']: data_actuator['pk']}
elif data_ve['load'] is not None: # energy related sensor: LOAD
data_load = data_ve['load']
load_id = data_load['pk']
# get the sensor linked to this LOAD entity
list_of_sensors = get_energy_entity_sensors(load_id, EMS_CATEGORY_ENTITY_LOAD)
elif data_ve['storage'] is not None: # energy related sensor: STORAGE
data_stor = data_ve['storage']
stor_id = data_stor['pk']
# get the sensor linked to this STORAGE entity
list_of_sensors = get_energy_entity_sensors(stor_id, EMS_CATEGORY_ENTITY_STORAGE)
elif data_ve['generator'] is not None: # energy related sensor: GENERATOR
data_gen = data_ve['generator']
gen_id = data_gen['pk']
# get the sensor linked to this GENERATION entity
list_of_sensors = get_energy_entity_sensors(gen_id, EMS_CATEGORY_ENTITY_GENERATION)
if list_of_sensors is None:
print("Couldn't associate sensor to: {0} ".format(ve))
self.ve_to_bms_mapping[ref] = list_of_sensors
# From the BMS ID to the vE ID
list_bms_entities = {'load': EMS_CATEGORY_ENTITY_LOAD,
'storage': EMS_CATEGORY_ENTITY_STORAGE,
'generator': EMS_CATEGORY_ENTITY_GENERATION,
'sensor': 'sensor',
'actuator': 'actuator'}
for type_e in list_bms_entities.values():
self.bms_to_ve_mapping[type_e] = {}
for ve in list_of_vEntities:
ve_id = ve['pk']
data_ve = ve['fields']
for type_e in list_bms_entities.keys():
if data_ve[type_e] is not None:
bms_id = data_ve[type_e]['pk']
self.bms_to_ve_mapping[list_bms_entities[type_e]][bms_id] = ve_id
break
### @ Message receiving and sending
def get_in_message(self, recv_timeout=0):
""" Receive multipart message from the interface """
if RT_EVENT_PROTOCOL == RT_EVENT_PROTOCOL_ZMQ:
recv_timeout *= 1000
socks = dict(self.socket_poller.poll(timeout=recv_timeout))
if socks.get(self.rtevent_sub) == zmq.POLLIN:
return self.rtevent_sub.recv_json()
else:
return None
elif RT_EVENT_PROTOCOL == RT_EVENT_PROTOCOL_WEBSOCKET:
#TODO
pass
def send_downstream_event(self, ems_msg):
# In case of SIMULATION with the vEngine, translate from BMS ID to vE ID
if RT_EVENT_PROTOCOL == RT_EVENT_PROTOCOL_ZMQ and EMS_MODE == EMS_MODE_VE_SIMULATION:
ve_ref = None
msg_payload = {}
if ems_msg[EMS_DOWNSTREAM_MSG_TYPE] == EMS_DOWNSTREAM_ENERGY_ENTITY_COMMAND: # generate a command understood by the downstream entities
# Transform {bms_type, bms_id} into a vE ID
bms_id = ems_msg[EMS_DOWNSTREAM_MSG_ENTITY_ID]
bms_type = ems_msg[EMS_DOWNSTREAM_MSG_ENTITY_TYPE]
if bms_type not in self.bms_to_ve_mapping or bms_id not in self.bms_to_ve_mapping[bms_type]:
print("couldn't associate BMS ID {0} of type '{1}' with any vE ID".format(bms_id, bms_type))
return
ve_ref = self.bms_to_ve_mapping[bms_type][bms_id]
# Generate the command
msg_payload = {}
if ems_msg[EMS_DOWNSTREAM_ENERGY_ENTITY_COMMAND_TYPE] == EMS_ENERGY_power: # new power set point
type_msg = None
setting_msg = None
value = setting_msg = ems_msg[EMS_DOWNSTREAM_MSG_PAYLOAD][EMS_ENERGY_power]
if bms_type == EMS_CATEGORY_ENTITY_STORAGE:
type_msg = 'NEW_P'
setting_msg = value
elif bms_type == EMS_CATEGORY_ENTITY_LOAD:
# Retrieve the load profile structure
state = 'ON'
if value == 0:
state = 'OFF'
type_msg = "NEW_MODE"
setting_msg = {'State': state, 'Param': None, 'Quantity': value}
msg_payload = RTEventInterface.generate_command(ve_ref,
type_msg,
setting_msg)
elif ems_msg[EMS_DOWNSTREAM_MSG_TYPE] == EMS_DOWNSTREAM_SIMU_READY:
ve_ref = SIMULATION_VMID_ADDRESS
msg_payload = RTEventInterface.generate_ready_signal()
msg_ve = dict()
msg_ve['recipient'] = ve_ref
msg_ve['PAYLOAD'] = json.dumps(msg_payload)
self.rtevent_pub.send_json(msg_ve)
print("EMS sent downstream: {0}".format(msg_ve))
elif RT_EVENT_PROTOCOL == RT_EVENT_PROTOCOL_WEBSOCKET:
#TODO
pass
### @ Message decoding/encoding
def process_in_event(self, in_msg):
"""Interpret the message coming from the interface and update the appropriate structures"""
ems_msg_list = []
#print("RT interface processing {0}".format(in_msg))
if EMS_MODE == EMS_MODE_VE_SIMULATION:
msg_type = in_msg["MSG_TYPE"]
if msg_type == 'simulation_coordination':
data = in_msg["PAYLOAD"]
ems_msg = dict()
ems_msg[RT_EVENT_MSG_TYPE_KEY] = RT_EVENT_MSG_TYPE_SIMULATION_SYNQ
ems_msg[RT_EVENT_SIMULATION_PARAM_KEY] = data
return [ems_msg] # already well formatted
else:
ems_msg_type = None
if msg_type == "MEASURE" or msg_type == "STATUS":
ems_msg_type = RT_EVENT_MSG_TYPE_MEASURE_STATUS
ems_sender_type = None
if msg_type == "MEASURE":
ems_sender_type = RT_EVENT_SENDER_TYPE_SENSOR
elif msg_type == "STATUS":
ems_sender_type = RT_EVENT_SENDER_TYPE_ACTUATOR
data = in_msg["PAYLOAD"]
if data["Ref"] not in self.ve_to_bms_mapping or ems_msg_type is None or ems_sender_type is None:
return {}
ems_sender_id_list = self.ve_to_bms_mapping[data["Ref"]]
if ems_sender_id_list is None:
return {}
payload = data["Value"]["Quantity"]
for quant_type, value in payload.items():
ems_msg = dict()
ems_msg[RT_EVENT_MSG_TYPE_KEY] = ems_msg_type
ems_msg[RT_EVENT_SENDER_TYPE] = ems_sender_type
if quant_type not in DB_KEY_EMS_SENSORS_CONVERSION:
continue # The EMS doesn't know this type of message
ems_msg[RT_EVENT_QUANTITY_TYPE] = DB_KEY_EMS_SENSORS_CONVERSION[quant_type]
if DB_KEY_EMS_SENSORS_CONVERSION[quant_type] not in ems_sender_id_list.keys():
continue # The EMS doesn't know this quantity for this sender sender
ems_sender_id = ems_sender_id_list[ems_msg[RT_EVENT_QUANTITY_TYPE]]
ems_msg[RT_EVENT_SENDER_ID] = ems_sender_id
ems_msg[RT_EVENT_VALUE] = value
ems_msg_list.append(ems_msg)
return ems_msg_list
### 'NORMAL' BMS message ###
ems_msg = {}
# The ID of the sender
if "ID" in in_msg:
ems_msg[RT_EVENT_SENDER_ID] = in_msg["ID"]
else:
ems_msg[RT_EVENT_SENDER_ID] = None
# Simulation coordination:
type_msg = None
if "MSG_TYPE" in in_msg and in_msg["MSG_TYPE"] == RT_EVENT_COORDINATION_MSG_TYPE:
type_msg = RT_EVENT_MSG_TYPE_SIMULATION_SYNQ
# The PAYLOAD will define the type of message
if "PAYLOAD" in in_msg:
payload = in_msg['PAYLOAD']
if ("MSG_CLASS" in payload and payload["MSG_CLASS"] == "MEASURE") \
or ("Type" in payload and payload["Type"] == "MEASURE"):
ems_msg[RT_EVENT_MSG_TYPE_KEY] = RT_EVENT_MSG_TYPE_MEASURE_STATUS
ems_msg[RT_EVENT_SENDER_TYPE] = 'sensor'
ems_msg[RT_EVENT_QUANTITY_TYPE] = DB_KEY_EMS_SENSORS_CONVERSION[payload["QUANTITY"]]
ems_msg[RT_EVENT_VALUE] = payload["VALUE"]
elif ("MSG_CLASS" in payload and payload["MSG_CLASS"] == "STATUS") \
or ("Type" in payload and payload["Type"] == "STATUS"):
ems_msg[RT_EVENT_MSG_TYPE_KEY] = RT_EVENT_MSG_TYPE_MEASURE_STATUS
ems_msg[RT_EVENT_SENDER_TYPE] = 'actuator'
ems_msg[RT_EVENT_QUANTITY_TYPE] = DB_KEY_EMS_SENSORS_CONVERSION[payload["QUANTITY"]]
ems_msg[RT_EVENT_VALUE] = payload["VALUE"]
else:
pass
ems_msg[RT_EVENT_MSG_TYPE_KEY] = type_msg
return [ems_msg]
@staticmethod
def generate_command(target, type_action, setting):
"""
TODO
:param target:
:param type_action:
:param setting:
:return:
"""
ret = dict()
ret['MID_ID'] = SIMULATION_BMS_MID_ID
ret['REF'] = target
ret['MSG_TYPE'] = 'Command'
ret['PAYLOAD'] = dict()
ret['PAYLOAD']['Type'] = type_action
ret['PAYLOAD']['Setting'] = setting
ret['PAYLOAD']['Time'] = time.time()
return ret
@staticmethod
def generate_ready_signal():
return {"MSG_TYPE": "coord_simu_ready", "CONTENT": {}}
c4science · Help