AgentDevicePLC.py
3.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
#!/usr/bin/env python3
import sys,os
pwd = os.environ['PROJECT_ROOT_PATH']
if pwd not in sys.path:
sys.path.append(pwd)
from src.core.pyros_django.agent.Agent import Agent, build_agent, log
from src.core.pyros_django.devices.PLC import PLCController
from src.core.pyros_django.monitoring.plc_checker import PlcChecker
from common.models import *
from config.pyros.config_pyros import ConfigPyros
class AgentDevicePLC(Agent):
# FOR TEST ONLY
# Run this agent in simulator mode
TEST_MODE = False
# Run the assertion tests at the end
TEST_WITH_FINAL_TEST = True
TEST_MAX_DURATION_SEC = None
#TEST_MAX_DURATION_SEC = 120
plcController = PLCController()
print ("AGENT ENV: config PLC is (ip={}, port={})".format(plcController.ip, plcController.port))
plc_checker = PlcChecker()
log.debug("PLC instanciated")
time_history_minutes = 4
# new config (obsconfig)
def __init__(self, name:str=None):
if name is None:
name = self.__class__.__name__
super().__init__()
self.device_name = self.get_config().get_device_for_agent(self.get_config().unit_name, self.name).get("name")
self.serial_number = self.get_config().get_device_for_agent(self.get_config().unit_name, self.name).get("sn")
self.output_data = self.get_config().get_output_data_device(self.device_name)
# @override
def _init(self):
super()._init()
log.debug("end init()")
# --- Set the mode according the startmode value
##agent_alias = self.__class__.__name__
##self.set_mode_from_config(agent_alias)
# @override
def do_log(self):
super().do_log()
# @override
# previous name of function : routine_process
# Note : in Agent.py, routine_process_body seems to be the main function of routine of the agent
# We need to override routine_process_body and not routine_process
def _routine_process_iter_end_body(self):
log.debug("in routine_process_body()")
print("TODO: we recycle code")
status_plc = self.plcController.getStatus()
if self.parseNewStatus(status_plc):
self.saveSensors()
#self.saveInternalMonitoring()
def parseNewStatus(self,status_plc):
# """ PM 20181009 parse new status for config
# Find return string "plc_status" positin within status_plc
if status_plc.find('PLC_STATUS') >= 0:
self.plc_checker.chk_config(status_plc)
return True
return False
def saveSensors(self):
for sensor in self.output_data:
sensor_monitoring_name = self.output_data[sensor]["monitoring_name"]
if sensor_monitoring_name in self.plc_checker.monitoring_names.keys():
value = self.plc_checker.get_sensor(sensor)
sensors = Sensors_data_last_value.objects.create(
value=value,
name=sensor_monitoring_name,
key=sensor,
model=self.device_name,
serial_number=self.serial_number
)
sensors.save()
# We don't have an history for weatherwatch
log.debug("saved weather")
if __name__ == "__main__":
# with thread
RUN_IN_THREAD=True
# with process
#RUN_IN_THREAD=False
args = parse_args(sys.argv[1:])
agent = build_agent(AgentDevicePLC,param_constr=args)
'''
TEST_MODE, configfile = extract_parameters()
agent = AgentM("AgentM", configfile, RUN_IN_THREAD)
agent.setSimulatorMode(TEST_MODE)
'''
print(agent)
agent.run()