完整的数据库操作

hfv/Dockerfile

 FROM opengn.org:32333/easypi/alpine-arm
 MAINTAINER joliu<joliu@s-an.org>
 ENV TIME_ZONE Asia/Shanghai
-RUN apk --update add python3 py3-pip tzdata
-RUN pip3 install pymysql -i https://pypi.douban.com/simple
+RUN apk --update add python3 py3-pip tzdata make cmake gcc g++ gfortran python3-dev
+RUN pip3 install pymysql cython numpy -i https://pypi.douban.com/simple
 RUN echo "${TIME_ZONE}" > /etc/timezone \
 && ln -sf /usr/share/zoneinfo/${TIME_ZONE} /etc/localtime 
 
@@ -10,6 +10,9 @@ WORKDIR /data
 COPY listenSer.py /data
 COPY workProcess.py /data
 COPY testDevice.py /data
+COPY getData.py /data
+COPY resolveMatrix.py /data
+COPY controllMatrix.py /data
 COPY start.sh /data
 
 CMD sh /data/start.sh

hfv/controllMatrix.py

@@ -15,8 +15,10 @@ import time
 # ~~~~~~~~~~~~~~~~~~~inputDB操作函数~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 # 更新任务执行控制表
 def updateDeviceTask(data, dstIP, ctime):
-    sql = "select * from decidedestination"
+    sql = "select * from decidedestination where dst='%s'" % dstIP
     (status, output) = sendToDB(sql)
+    print(output)
+    print('***********',output)
     if status == -1:
         return (status, output)
     if output == []:
@@ -34,10 +36,12 @@ def getDeviceTask():
     (status, output) = sendToDB(sql)
     if status == -1:
         return (status, output)
-    return (status, output[0])
+    # print(output)
+    return (status, output)
 
 
 # ~~~~~~~~~~~~~~~~~~~dataCach操作函数~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
 # 更新设备输入数据信息
 def updateDataCach(device, decideValue):
     # 检查是否存在该设备在预接收的列表中，如果不存在则丢弃
@@ -71,9 +75,51 @@ def getDataFromDataCach(device):
     sql = "select data from datacach where deviceid='%s'" % device
     return sendToDB(sql)
 
+# 判断是否存在相应的设备数据缓存信息
+def existDevice(device):
+    (status, output) = getDataFromDataCach(device)
+    print(device, output)
+    if status == -1:
+        retrun (status, output)
+    if output == []:
+        return (1, False)
+    else:
+        return (1, True) 
 
 # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 # 以下都是对output的数据库，即存储任务矩阵的数据库的操作
+# 获取设备列表
+def getDeviceListFromDB():
+    sql = "select devicelist from resulovetable"
+    (status, output) = sendToDB(sql)
+    if status == -1:
+        return (status, output)
+    if output == []:
+        return (-1, 'no deviceList')
+    return (status, output[0][0])
+
+
+# 获取任务执行状态
+def getTaskStatus():
+    sql = "select status from resulovetable"
+    (status, output) = sendToDB(sql)
+    print(output)
+    if status == -1:
+        return (status, output)
+    if output == []:
+        return (-1, 'no deviceList')
+    return (status, output[0][0])
+
+
+# 更新任务执行状态
+def updateTaskStatus(taskStatus):
+    (status, output) = getTaskStatus()
+    if status == -1:
+        return (status, output)
+    sql = 'update resulovetable set status=%d' % taskStatus
+    return sendToDB(sql)
+
+
 # 插入任务到数据库
 def insertDB(taskMatrix, inputTypeList, deviceList, status, ctime):
     sql = "insert into resulovetable \
@@ -107,7 +153,7 @@ def getTaskFromDB():
     # 查询最新一条可用任务
     sql = 'select * from resulovetable order by id desc limit 0,1'
     (status, output) = sendToDB(sql)
-    return output[0]
+    return output
 
 def getTaskFromDBByID(id):
     # 从数据库中获取数据
@@ -117,7 +163,7 @@ def getTaskFromDBByID(id):
 
 def getValueByNodeID(nodeid):
     # 获取devicelist
-    (id, inputTask, inputTypeList, status, deviceList, ctime) = getTaskFromDB()
+    (id, inputTask, inputTypeList, status, deviceList, ctime) = getTaskFromDB()[0]
     device = json.loads(deviceList)[nodeid]
     if device == -1:
         return (-1, "not a device")
@@ -132,7 +178,7 @@ def getValueByNodeID(nodeid):
 
 # 获取循环时间
 def getCircleTime():
-    data = getTaskFromDB()
+    data = getTaskFromDB()[0]
     if data == ():
         return 5
     return data[5]
@@ -180,7 +226,7 @@ if __name__ == '__main__':
     for i in range(len(data[0])):
         insertDataIntoDataCach(data[0][i])
         updateDataCach(data[0][i], data[1][i])
-    showDB()
-    (id, inputTask, inputTypeList, status, deviceList, ctime) = getTaskFromDB()
+    #showDB()
+    (id, inputTask, inputTypeList, status, deviceList, ctime) = getTaskFromDB()[0]
     #print(inputTask)
     print(getValueByNodeID(6))

hfv/listenSer.py

@@ -29,7 +29,7 @@ from controllMatrix import *
 
 # 两种控制模式，controller:控制器写入控制命令，device：接收其他传感器控制命令
 controlModeList = ['controller', 'device']
-controlMethodList = ['addInput','addOutput', 'rm', 'clear', 'period', 'show']
+controlMethodList = ['addInput','addOutput', 'rm', 'clear', 'period', 'show', 'start', 'stop']
 
 class ThreadedTCPRequestHandler(socketserver.BaseRequestHandler):
     '''
@@ -70,18 +70,20 @@ class ThreadedTCPRequestHandler(socketserver.BaseRequestHandler):
                     # 匹配控制指令做出相应操作
                     (status, output) = executeCommand(command, message[2:])
                     print(message[2:])
-                    #(status , output) = (1, message[2:])
 
             # 监听来自device hfv模块的控制请求
             elif controlMode == "device":
-                command = message[1]
+                deviceID = message[1]
+                data = message[2]
                 # 发送控制请求
-                (status, output) = sendCommandToDevice(command)
-                # (status, output) = (1, "test")
+                # (status, output) = sendCommandToDevice(command)
+                (status, output) = updateDataCach(deviceID, int(data))
+                print(deviceID, data)
+                #(status, output) = (1, "test")
             else:
                 pass
         else:
-            print("illegal controlMode")
+            print("illegal controlMode!")
             (status, output) = (-1, 'illegal controlMode')
             errorFlag = True
 
@@ -154,11 +156,24 @@ def executeCommand(command, information):
              circleTime) = information
         return insertDB(taskMatrixJOSN, deviceTypeListJOSN, deviceListJOSN,\
                             int(taskStatus), int(circleTime))
+    elif command == 'addCach':
+        data = information[0]
+        data = json.loads(data)
+        for i in range(len(data[0])):
+            (status, output) = insertDataIntoDataCach(data[0][i])
+            (status, output) = updateDataCach(data[0][i], data[1][i])
+        return (status, output)
+        
     elif command == 'show':
         DBName = information[0]
         return showDatabase(DBName)
     elif command == 'clear':
         return clearDB()
+    elif command == 'start':
+        return updateTaskStatus(1)
+    elif command == 'stop':
+        return updateTaskStatus(0)
+
 
 # 创建数据库
 def createDB():

hfv/resolveMatrix.py

@@ -51,6 +51,35 @@ def resolveMatrix(inputTask, inputTypeList, outputNode):
 
 
 def checkMatrix(inputTask, inputTypeList): 
+    # 验证数据数据库数据是否否和规定
+    (status, output) = getDeviceListFromDB()
+    if status == -1:
+        print(output)
+        return False
+    (getTaskStatusStatus, taskStatus) = getTaskStatus()
+    if getTaskStatusStatus == -1:
+        print(taskStatus)
+    if taskStatus == 0:
+        print("Task has been stopped!")
+        return False
+    
+    deviceList = json.loads(output)
+    for device in deviceList:
+        if device == -1:
+            continue
+        if device[:5] == 'delay':
+            continue
+        if device[:6] == 'switch':
+            # 临时解决方案，把switch都作为输出屏蔽掉
+            continue
+        (runStatus, deviceStatus) = existDevice(device)
+        if runStatus == -1:
+            print(deviceStatus)
+            return False
+        if not deviceStatus:
+            print("need to set datacach database")
+            return False
+        
     # 测试矩阵是否符合要求
     length = inputTask.shape[0]
     zeroList = np.zeros(length)
@@ -84,8 +113,10 @@ def checkMatrix(inputTask, inputTypeList):
 
 def runTask():
     # 输入各个节点的状态信息
-
-    (id, inputTask, inputTypeList, status, deviceList, ctime) = getTaskFromDB()
+    data = getTaskFromDB()
+    if data == []:
+        return (-1, 'No Task')
+    (id, inputTask, inputTypeList, status, deviceList, ctime) = data[0]
 
     inputTask = np.array(json.loads(inputTask))
     inputTypeList = np.array(json.loads(inputTypeList))