| [volttron-dev] Custom Controller Agent Causes Modbus Read Error |
|
Hello Volttron Team, Background: I am using Volttron 8.1.3 on a Raspberry Pi to coordinate the operation / testing of 2 devices over Modbus:
Using the Listener Agent I can see that the platform.driver is reading all of my modbus points perfectly every 5 seconds (see the attached volttron_Without_Controller.log). I have a simple custom agent, enervenue_ctrlagent (see the “agent.txt” file, changed from “agent.py” for attaching), that is subscribed to the "devices/STAC/EnervenueBMS" topic and runs a simple read from the battery, decide what to do,
and write to the inverter loop every 5 seconds when new "devices/STAC/EnervenueBMS" data comes in. The writing is just to a single “Output Power Command” register on the inverter.
The Problem: When enervenue_ctrlagent is installed and started, modbus reads of the inverter begin to fail. The controller’s attempts to write power setpoints are successful, but I start to receive this modbus_tk error (see the attached volttron_With_Controller.log):
Expected Behavior: I was expecting that my controller agent should be able to run in tandem with the platform agent with no conflicts between reading and writing, but something funny is happening that I don’t understand.
Any and all help is greatly appreciated! Joe Thompson Engineer / Scientist Electric Power Research Institute Energy Storage and Distributed Generation
(912) 663-3407 |
Attachment:
volttron_Without_Controller.log
Description: volttron_Without_Controller.log
"""
Agent documentation goes here.
"""
__docformat__ = 'reStructuredText'
import logging
import sys
from volttron.platform.agent import utils
from volttron.platform.vip.agent import Agent, Core, RPC
from volttron.platform.agent.utils import format_timestamp, get_aware_utc_now
import pandas as pd
_log = logging.getLogger(__name__)
utils.setup_logging()
__version__ = "0.1"
def enervenue_ctrl(config_path, **kwargs):
"""
Parses the Agent configuration and returns an instance of
the agent created using that configuration.
:param config_path: Path to a configuration file.
:type config_path: str
:returns: EnervenueCtrl
:rtype: EnervenueCtrl
"""
try:
config = utils.load_config(config_path)
except Exception:
config = {}
if not config:
_log.info("Using Agent defaults for starting configuration.")
setting1 = int(config.get('setting1', 1))
setting2 = config.get('setting2', "some/random/topic")
sched_path = config['control_schedule_path']
return EnervenueCtrl(sched_path, setting1, setting2, **kwargs)
class EnervenueCtrl(Agent):
"""
Document agent constructor here.
"""
def __init__(self, sched_path, setting1=1, setting2="some/random/topic", **kwargs):
super(EnervenueCtrl, self).__init__(**kwargs)
_log.debug("vip_identity: " + self.core.identity)
self.agent_id = "enervenue_sched_cntl" #enervenue_ctrl
self.sched_path = sched_path
self.setting1 = setting1
self.setting2 = setting2
self.bms_write_device = 'STAC/EnervenueBMS'
self.inverter_write_device = 'STAC/DynapowerInverter'
self.inverter_rated_kw = 124.7
# Configure the actual modbus writes and format the write request
self.dynapower_control = {
'Fault Reset': 0, # 0: No Action 1: Reset All Latched Alarms/Faults
'Operation Mode Select': 3, # 1: Idle State 3: Grid-Tied Mode
'Output Power Command': 0, # Percentage of rated power (124.7kW), negative = charge
'PCS Set Operation': 0, # 0: No Action 1: Start PCS Operation 2: Stop PCS Operation
}
self.default_config = {
"control_schedule_path": sched_path,
"setting1": setting1,
"setting2": setting2
}
# Set a default configuration to ensure that self.configure is called immediately to setup
# the agent.
self.vip.config.set_default("config", self.default_config)
# Hook self.configure up to changes to the configuration file "config".
self.vip.config.subscribe(self.configure, actions=["NEW", "UPDATE"], pattern="config")
def configure(self, config_name, action, contents):
"""
Called after the Agent has connected to the message bus. If a configuration exists at startup
this will be called before onstart.
Is called every time the configuration in the store changes.
"""
config = self.default_config.copy()
config.update(contents)
_log.debug("Configuring Agent")
try:
sched_path = config["control_schedule_path"]
setting1 = int(config["setting1"])
setting2 = str(config["setting2"])
except ValueError as e:
_log.error("ERROR PROCESSING CONFIGURATION: {}".format(e))
return
self.setting1 = setting1
self.setting2 = setting2
self.sched_path = sched_path
self.system_topic = "devices/STAC/EnervenueBMS"
self.sched = pd.read_excel(self.sched_path)
self.sched_step_ind = 0
self.parse_control_step()
self.timed_step_end_utc = None if pd.isna(self.duration) else pd.Timestamp.utcnow() + pd.Timedelta(seconds=self.duration)
self._create_subscriptions(self.system_topic)
# self._create_subscriptions(self.setting2)
### SUBSCRIBE TO BESS AND INVERTER POINTS. GO THROUGH CONTROL LOOP AT EACH READ
def _create_subscriptions(self, topic):
"""
Unsubscribe from all pub/sub topics and create a subscription to a topic in the configuration which triggers
the _handle_publish callback
"""
self.vip.pubsub.unsubscribe("pubsub", None, None)
# subscribe to all read data from the BMS
self.vip.pubsub.subscribe(peer='pubsub',
prefix=topic,
callback=self._handle_publish)
def parse_control_step(self):
# get details for the current control step
self.ctrl_step = self.sched.loc[self.sched_step_ind]
self.step_name = self.ctrl_step['Step']
self.target_soc = self.ctrl_step['Target SoC (%)']
self.power_cmd = self.ctrl_step['Power (kW)\n+ = discharge']
self.duration = self.ctrl_step['Duration (s)']
self.action = 'discharge' if self.power_cmd > 0 else 'charge' if self.power_cmd < 0 else 'idle'
def _handle_publish(self, peer, sender, bus, topic, headers, message):
"""
Callback triggered by the subscription setup using the topic from the agent's config file
For now we can write out all the control logic here
"""
current_soc = float(message[0]['EnerStation SOC'])
connected_strings = sum([message[0]['String %d Relay status' % s] for s in range(1, 4)])
# adjust the power request given the number of strings online
try:
power_command = (self.power_cmd / 3) * connected_strings
except ZeroDivisionError as e:
do_not_proceed = True
_log.info("No Strings are connected. Stopping the inverter.")
self.write_to_device(
device=self.inverter_write_device,
write_values={
'Output Power Command': 0, # Percentage of rated power (124.7kW), negative = charge
'PCS Set Operation': 2, # 0: No Action 1: Start PCS Operation 2: Stop PCS Operation
}
)
# convert power command into percent of rated power
power_command_pct = (power_command / self.inverter_rated_kw) * 100
# if the current step name is not an integer then send an idle command
try:
int(self.ctrl_step.name)
do_not_proceed = False
except ValueError as e:
do_not_proceed = True
_log.info("Schedule was completed. Stopping the inverter.")
self.write_to_device(
device=self.inverter_write_device,
write_values={
'Output Power Command': 0, # Percentage of rated power (124.7kW), negative = charge
'PCS Set Operation': 2, # 0: No Action 1: Start PCS Operation 2: Stop PCS Operation
}
)
# check whether we are in a timed / duration step
if self.timed_step_end_utc is not None: # we are within a timed step
if pd.Timestamp.utcnow() < self.timed_step_end_utc:
# send write command that corresponds to this step
self.write_to_device(
device=self.inverter_write_device,
write_values={
'Output Power Command': power_command_pct, # Percentage of rated power (124.7kW), negative = charge
}
)
# do not proceed with the rest of control logic
# wait for next read from battery
return
else: # enough time has passed move on to the next step
self.timed_step_end_utc = None
self.sched_step_ind += 1
self.parse_control_step()
if pd.isna(self.duration): # then this is a power / SoC command
if (self.action == 'charge') & (current_soc < self.target_soc):
_log.info("Sending charge command of %0.1fkW" % power_command)
self.write_to_device(
device=self.inverter_write_device,
write_values={
'Output Power Command': -20 #power_command_pct, # Percentage of rated power (124.7kW), negative = charge
}
)
elif (self.action == 'discharge') & (current_soc > self.target_soc):
_log.info("Sending discharge command of %0.1fkW" % power_command)
self.write_to_device(
device=self.inverter_write_device,
write_values={
'Output Power Command': power_command_pct, # Percentage of rated power (124.7kW), negative = charge
}
)
else: # move on to the next step and execute that step
self.sched_step_ind += 1
self.parse_control_step()
self.timed_step_end_utc = None if pd.isna(self.duration) else pd.Timestamp.utcnow() + pd.Timedelta(seconds=self.duration)
_log.info("Moving on to next step with power of %0.1fkW" % power_command)
self.write_to_device(
device=self.inverter_write_device,
write_values={
'Output Power Command': power_command_pct, # Percentage of rated power (124.7kW), negative = charge
}
)
else: # this must be a duration command
self.timed_step_end_utc = None if pd.isna(self.duration) else pd.Timestamp.utcnow() + pd.Timedelta(seconds=self.duration)
self.write_to_device(
device=self.inverter_write_device,
write_values={
'Output Power Command': power_command_pct, # Percentage of rated power (124.7kW), negative = charge
}
)
def write_to_device(self, device, write_values):
"""
Write some points to a device
"""
# Create a start and end timestep to serve as the times we reserve to communicate with the device
_now = get_aware_utc_now()
str_now = format_timestamp(_now)
_end = _now + pd.Timedelta(seconds=3600)
str_end = format_timestamp(_end)
# Wrap the timestamps and device topic (used by the Actuator to identify the device) into an actuator request
schedule_request = [[device, str_now, str_end]]
# Use a remote procedure call to ask the actuator to schedule us some time on the device
_log.info("schedule_request {}".format(schedule_request))
result = self.vip.rpc.call(
'platform.actuator', 'request_new_schedule', self.core.identity, 'my_test', 'HIGH', schedule_request).get(
timeout=4)
# start by creating our topic_values
topic_values = []
for point, value in write_values.items():
# create a (topic, value) tuple and add it to our topic values
topic_values.append((device + '/' + point, value))
# Now use another RPC call to ask the actuator to set the point during the scheduled time
_log.info("Modbus points to write: {}".format(topic_values))
result = self.vip.rpc.call(
'platform.actuator', 'set_multiple_points', self.core.identity, topic_values).get(
timeout=4)
_log.info("Modbus write response: {}".format(result))
@Core.receiver("onstart")
def onstart(self, sender, **kwargs):
"""
This is method is called once the Agent has successfully connected to the platform.
This is a good place to setup subscriptions if they are not dynamic or
do any other startup activities that require a connection to the message bus.
Called after any configurations methods that are called at startup.
Usually not needed if using the configuration store.
"""
# Example publish to pubsub
# self.vip.pubsub.publish('pubsub', "some/random/topic", message="HI!")
_log.info("Schedule Controller started")
# Example RPC call
# self.vip.rpc.call("some_agent", "some_method", arg1, arg2)
pass
@Core.receiver("onstop")
def onstop(self, sender, **kwargs):
"""
This method is called when the Agent is about to shutdown, but before it disconnects from
the message bus.
"""
self.write_step(power=0)
pass
@RPC.export
def rpc_method(self, arg1, arg2, kwarg1=None, kwarg2=None):
"""
RPC method
May be called from another agent via self.core.rpc.call
"""
return self.setting1 + arg1 - arg2
def main():
"""Main method called to start the agent."""
utils.vip_main(enervenue_ctrl,
version=__version__)
if __name__ == '__main__':
# Entry point for script
try:
sys.exit(main())
except KeyboardInterrupt:
pass
Attachment:
volttron_With_Controller.log
Description: volttron_With_Controller.log