Skip to content

O
OpenCPS
Commit d11e5a0c authored by joliu's avatar joliu
Browse files
Options

变更为同时处理多任务

parent 454598de
Show whitespace changes
Inline Side-by-side
Showing with 88 additions and 78 deletions
hfv/workProcess.py
View file @ d11e5a0c
......@@ -4,7 +4,8 @@
* 1. 检查任务队列是否存在任务 *
* 2. 从任务队列中取任务并执行 *
* 3. 发送控制命令到指定的智能体 *
* 4. 未来或许可以根据配置的智能体参数， *
* 4. 读取全部任务队列中任务，并依次执行 *
* 5. 未来或许可以根据配置的智能体参数， *
* 动态修改智能体的计算资源等 *
* *
* author： joliu<joliu@s-an.org> *
......@@ -20,7 +21,7 @@ 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)
......@@ -37,30 +38,12 @@ 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)
......@@ -70,24 +53,47 @@ def mainWhileProcess(input_ctime):
(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/hfv2/workProcess.py
View file @ d11e5a0c
......@@ -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("cycle time :" + str(input_ctime))
time.sleep(input_ctime)
(status, output) = doTask()
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(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())
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