Skip to content

O
OpenCPS
Commit 8d21dacd authored by joliu's avatar joliu Committed by android
Browse files
Options

Merge branch 'multi-task' of joliu/opencps-agent into master

parents d3bfba7f bca730ae
Hide whitespace changes
Inline Side-by-side
Showing with 126 additions and 81 deletions
README.md
View file @ 8d21dacd
......@@ -24,6 +24,7 @@
##### 2018-3-26
1. 完成在Docker上的第一版代码测试（已修复BUG）
2. 搭建了一套完整的本地测试系统（包括dht102和switch104）
3. 搭建同时执行多条任务代码
## 版本说明
......@@ -42,7 +43,7 @@
注： 控制器命令目前由`./tests/testListenSer.py`代为执行
### 版本目前存在问题
1. 任务队列中只能存在一条任务
1. 任务队列中只能存在一条任务 (已解决)
### 未来要支持的功能
......
hfv/workProcess.py
View file @ 8d21dacd
......@@ -4,7 +4,8 @@
* 1. 检查任务队列是否存在任务 *
* 2. 从任务队列中取任务并执行 *
* 3. 发送控制命令到指定的智能体 *
* 4. 未来或许可以根据配置的智能体参数， *
* 4. 读取全部任务队列中任务，并依次执行 *
* 5. 未来或许可以根据配置的智能体参数， *
* 动态修改智能体的计算资源等 *
* *
* author： joliu<joliu@s-an.org> *
......@@ -20,11 +21,11 @@ import time
def findTask():
try:
sql = "select * from task LIMIT 1"
sql = "select * from task"
conn = sqlite3.connect("task.db")
cursor = conn.cursor()
cursor.execute(sql)
data = cursor.fetchone()
data = cursor.fetchall()
if data is None:
(status, output) = (-1, 'database is empty!')
else:
......@@ -37,57 +38,62 @@ def findTask():
return (status, output)
# 根据符号来比较两个数值的大小
def compare(signal, value1, value2):
if signal == '>':
return value1 > value2
elif signal == '<':
return value1 < value2
elif signal == '=':
return value1 == value2
else:
return False
def mainWhileProcess(input_ctime):
ctime = input_ctime
while True:
print(ctime)
time.sleep(ctime)
(status, output) = findTask()
if status == -1:
print(output)
# 如果数据库为空，或者错误，恢复初始设置
ctime = input_ctime
continue
# 根据任务队列中任务，依次执行
def doTask():
(status, outputs) = findTask()
if status != 1:
return (status, outputs)
for output in outputs:
(task, taskid, ctime) = output
if len(task.split(';')) != 2 or len(task.split(':')) != 3:
print("Error task: %s" % task)
continue
# 初步定义task字符串模式 eg: >30;192.168.1.1:3000:off
(condition, command) = task.split(';')
(ip, port, method) = command.split(':')
# 构建控制指令
# 构建控制命令
method = 'device&' + method
# 读取传感器数值
(status, output) = sendCommandToDevice(method)
# 针对dht11的数据格式做一个简单的数据预处理
output = output.split('&')[0]
(status, data) = sendCommandToDevice(method)
# 千杀的dht11，需要处理下数据
data = data.split('&')[0]
print(output)
if status == -1:
print("get device data failed! ip: %s, method: %s" % (ip, method))
continue
return (status, "get device data failed! ip: %s, method: %s" % (ip, method))
if compare(condition[0], float(output), float(condition[1:])):
if compare(condition[0], float(data), float(condition[1:])):
# 当结果为真，向目标传感器发出指令
(status, output) = sendBySocket(ip, int(port), method)
print(output)
(status, recvdata) = sendBySocket(ip, int(port), method)
print(recvdata)
time.sleep(ctime)
else:
pass
return (1, 'success')
# 根据符号来比较两个数值的大小
def compare(signal, value1, value2):
if signal == '>':
return value1 > value2
elif signal == '<':
return value1 < value2
elif signal == '=':
return value1 == value2
else:
return False
def mainWhileProcess(input_ctime):
while True:
print("cycle time :" + str(input_ctime))
time.sleep(input_ctime)
(status, output) = doTask()
print(output)
if __name__ == "__main__":
mainWhileProcess(5)
tests/hfv1/1.log
View file @ 8d21dacd
......@@ -138,3 +138,9 @@ Server loop running in thread:Thread-1
Program started
Server loop running in thread:Thread-1
....waiting for connection
Program started
Server loop running in thread:Thread-1
....waiting for connection
Program started
Server loop running in thread:Thread-1
....waiting for connection
tests/hfv1/README.md 0 → 100644
View file @ 8d21dacd
# 测试程序说明
继电器虚拟化模块
指定传感器ip地址为：192.168.12.42
指定传感器listenSer端口号为: 30001
## 使用说明
需要先启动nodemcu，待自动部署成功之后再另行进行测试。
tests/hfv1/task.db
View file @ 8d21dacd
No preview for this file type
tests/hfv2/1.log
View file @ 8d21dacd
......@@ -126,3 +126,9 @@ Program started
Program started
Server loop running in thread:Thread-1
....waiting for connection
Program started
Server loop running in thread:Thread-1
....waiting for connection
Program started
Server loop running in thread:Thread-1
....waiting for connection
tests/hfv2/README.md 0 → 100644
View file @ 8d21dacd
# 测试程序说明
DHT11虚拟化模块
指定传感器ip地址为：192.168.12.75
指定传感器listenSer端口号为: 30002
## 使用说明
需要先启动nodemcu，待自动部署成功之后再另行进行测试。
tests/hfv2/task.db
View file @ 8d21dacd
No preview for this file type
tests/hfv2/workProcess.py
View file @ 8d21dacd
......@@ -4,7 +4,8 @@
* 1. 检查任务队列是否存在任务 *
* 2. 从任务队列中取任务并执行 *
* 3. 发送控制命令到指定的智能体 *
* 4. 未来或许可以根据配置的智能体参数， *
* 4. 读取全部任务队列中任务，并依次执行 *
* 5. 未来或许可以根据配置的智能体参数， *
* 动态修改智能体的计算资源等 *
* *
* author： joliu<joliu@s-an.org> *
......@@ -26,11 +27,11 @@ def sendCommandToDevice(cmd):
def findTask():
try:
sql = "select * from task LIMIT 1"
sql = "select * from task"
conn = sqlite3.connect("task.db")
cursor = conn.cursor()
cursor.execute(sql)
data = cursor.fetchone()
data = cursor.fetchall()
if data is None:
(status, output) = (-1, 'database is empty!')
else:
......@@ -43,6 +44,43 @@ def findTask():
return (status, output)
# 根据任务队列中任务，依次执行
def doTask():
(status, outputs) = findTask()
if status != 1:
return (status, outputs)
for output in outputs:
(task, taskid, ctime) = output
if len(task.split(';')) != 2 or len(task.split(':')) != 3:
print("Error task: %s" % task)
continue
# 初步定义task字符串模式 eg: >30;192.168.1.1:3000:off
(condition, command) = task.split(';')
(ip, port, method) = command.split(':')
# 构建控制命令
method = 'device&' + method
# 读取传感器数值
(status, data) = sendCommandToDevice(method)
# 千杀的dht11，需要处理下数据
data = data.split('&')[0]
print(output)
if status == -1:
print("get device data failed! ip: %s, method: %s" % (ip, method))
return (status, "get device data failed! ip: %s, method: %s" % (ip, method))
if compare(condition[0], float(data), float(condition[1:])):
# 当结果为真，向目标传感器发出指令
(status, recvdata) = sendBySocket(ip, int(port), method)
print(recvdata)
time.sleep(ctime)
else:
pass
return (1, 'success')
# 通过socket发送信息
def sendBySocket(ip, port, cmd):
try:
......@@ -92,46 +130,12 @@ def compare(signal, value1, value2):
def mainWhileProcess(input_ctime):
ctime = input_ctime
while True:
print(ctime)
time.sleep(ctime)
(status, output) = findTask()
if status == -1:
print(output)
# 如果数据库为空，或者错误，恢复初始设置
ctime = input_ctime
continue
(task, taskid, ctime) = output
if len(task.split(';')) != 2 or len(task.split(':')) != 3:
print("Error task: %s" % task)
continue
# 初步定义task字符串模式 eg: >30;192.168.1.1:3000:off
(condition, command) = task.split(';')
(ip, port, method) = command.split(':')
# 构建控制命令
method = 'device&' + method
# 读取传感器数值
(status, output) = sendCommandToDevice(method)
# 千杀的dht11，需要处理下数据
output = output.split('&')[0]
print("cycle time :" + str(input_ctime))
time.sleep(input_ctime)
(status, output) = doTask()
print(output)
if status == -1:
print("get device data failed! ip: %s, method: %s" % (ip, method))
continue
if compare(condition[0], float(output), float(condition[1:])):
# 当结果为真，向目标传感器发出指令
(status, output) = sendBySocket(ip, int(port), method)
#print(output)
else:
pass
print(1212)
if __name__ == "__main__":
mainWhileProcess(5)
#print(doTask())
tests/testListenSer.py
View file @ 8d21dacd
......@@ -6,7 +6,7 @@ message = sys.argv[2]
# message = "controller&clear&niubi"
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("127.0.0.1", port))
s.connect(("192.168.12.19", port))
s.sendall(message.encode())
response = s.recv(1024).decode()
print(response)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment