thread repair

.env

@@ -3,7 +3,7 @@ hostName = 'https://212.224.86.112:8443'
 dbPath = './DbFuncs/sql.db'
 adPath = '/img/ad.mp4'
 
-screenX = 600
+screenX = 1000
 screenY = 900
 
 itemLength = 220

ad.py

@@ -38,6 +38,7 @@ def retrieve_layout(self, dt):
                 temp_file = path + os.environ.get('adPath')
                 utils.write_to_file(ad.content, temp_file)
 
+                # videoPlayerLayout = VideoPlayerLayout('./img/ad.mp4')
                 videoPlayerLayout = VideoPlayerLayout(temp_file)
                 videoPlayerLayout.manager = self.manager
                 videoPlayerLayout.bind(on_touch_up=videoPlayerLayout.on_video_touch_up)
@@ -100,17 +101,12 @@ def __init__(self, temp_file, **kwargs):
         super(VideoPlayerLayout, self).__init__(**kwargs)
         self.manager = None
         self.temp_file = temp_file
-
         # Create a VideoPlayer widget
-        self.player = Video(source=temp_file, state='play',
-                                  options={'eos': 'loop'})
-
-        # Add the VideoPlayer widget to the BoxLayout
+        self.player = Video(source=temp_file, state='play', options={'eos':'loop'})
         self.add_widget(self.player)
 
     def on_video_touch_up(self, video, touch):
         # Handle the video player touch up event
         if video.collide_point(*touch.pos):
             self.manager.current = 'List'
             # os.remove(self.temp_file)
-

api.py

@@ -17,32 +17,21 @@
 from dotenv import load_dotenv
 load_dotenv()
 from config.global_vars import global_ads, global_machines, global_produts
-from config.utils import lockList
-
+from config.utils import setThreadStatus, getThreadStatus, THREAD_INIT, THREAD_RUNNING, THREAD_STOPPING, THREAD_FINISHED
+from config.utils import getDBLock, DBLOCK_ADS, DBLOCK_MACHINE, DBLOCK_PRODUCT
+from websockets.exceptions import ConnectionClosed
 
 hostName = os.environ.get('hostName')
 # requestTimeStep = int(os.environ.get('requestTimeStep'))
 
-# def create_connect():
-#     ws = create_connection("wss://212.224.86.112:8443")
-#     print(ws.recv())
-#     print("Sending 'Hello, World'...")
-#     ws.send("Hello, World")
-#     print("Sent")
-#     print("Receiving...")
-#     result =  ws.recv()
-#     print("Received '%s'" % result)
-#     ws.close()
-
-# create_connect()
-
 def send_get_ads_info():
     url = "/api/machine/get_ads_info"
     # Send an HTTP GET request to a URL of your choice
     response = requests.post(hostName + url, verify=False)
 
-    if lockList[1]:
+    if getThreadStatus() == THREAD_STOPPING:
         return
+
     # Check the response status code
     if response.status_code == 200:
         responseData = response.json()   # {status : , message : , details : [{},]}
@@ -52,7 +41,7 @@ def send_get_ads_info():
         db.delete_ads()
         params = Ad(0, adData['type'], base64.b64decode(adData['content']))
         global_ads.append(params)
-        if lockList[1]:
+        if getThreadStatus() == THREAD_STOPPING:
             return
         db.insert_ads(params)
     else:
@@ -65,7 +54,7 @@ def send_get_machine_info():
     url = "/api/machine/get_machine_info"
     response = requests.post(hostName + url, verify=False)
 
-    if lockList[1]:
+    if getThreadStatus() == THREAD_STOPPING:
         return
 
     # Check the response status code
@@ -80,8 +69,8 @@ def send_get_machine_info():
 
             global_machines.append(params)
 
-            if lockList[1]:
-                break
+            if getThreadStatus() == THREAD_STOPPING:
+                return
             db.insert_machine(params)
     else:
         print(f"Request failed with status code: {response.status_code}")
@@ -91,7 +80,7 @@ def send_get_products_info():
 
     # Send an HTTP GET request to a URL of your choice
     response = requests.post(hostName + url, verify=False)
-    if lockList[1]:
+    if getThreadStatus() == THREAD_STOPPING:
         return
 
     # Check the response status code
@@ -104,7 +93,7 @@ def send_get_products_info():
             params = Product(0, item['itemno'], item['name'], base64.b64decode(item['thumbnail']), item['nicotine'], item['batterypack'],
                              item['tankvolumn'], item['price'], item['currency'], item['caution'], item['stock'])
             global_produts.append(params)
-            if lockList[1]:
+            if getThreadStatus() == THREAD_STOPPING:
                 break
             db.insert_product(params)
     else:
@@ -124,86 +113,111 @@ async def connect_to_server():
     ssl_context.verify_mode = ssl.CERT_NONE
 
     print("wss thread id", threading.get_native_id())
+
+    setThreadStatus(THREAD_RUNNING)
+
+    cnt = 0
 
     try:
-        async with websockets.connect('wss://212.224.86.112:8443', ssl = ssl_context) as websocket:
-            print('connected')
+        websocket = await websockets.connect('wss://212.224.86.112:8443', ssl = ssl_context)
+    except:
+        pass
+
+    while True: 
+        try:
+            if getThreadStatus() == THREAD_STOPPING:
+                break
+
+            if cnt % 50 != 0:
+                time.sleep(0.1)
+                cnt = (cnt+1) % 50
+                print('cnt', cnt)
+                continue
+
+            cnt = (cnt+1) % 50
+
+            # async with websockets.connect('wss://212.224.86.112:8443', ssl = ssl_context) as websocket:
+
+            if getThreadStatus() == THREAD_STOPPING:
+                break
+
             sendData = {'action': 'MachineConnect'}
             await websocket.send(json.dumps(sendData))
 
+            if getThreadStatus() == THREAD_STOPPING:
+                break
+
             # Receive data
             response = await websocket.recv()
             responseData = json.loads(response)
-
             print(f"Received data: {responseData}")
 
             machineConnectStatus = responseData['status']
             token = responseData['token']
-            cnt = 0
+
+            if getThreadStatus() == THREAD_STOPPING:
+                break                
 
             if machineConnectStatus == 'success':
-                while True: 
-                    if lockList[1]:
-                        break
-
-                    if cnt % 50 != 0:
-                        time.sleep(0.1)
-                        cnt = cnt+1
-                        continue
-
-                    cnt = cnt+1
-                    # print(f'stopConnncet: {stopConnect}')
-                    # if stopConnect:
-                    #     print('http_request_close')
-                    #     break
-
-                    statusData = {
-                        'action': "MachineSendStatus",
-                        'payload': {
-                            'serialno': "123-456-678",
-                            'temparature': "XXX",
-                            'token': token, 
-                        }
+                statusData = {
+                    'action': "MachineSendStatus",
+                    'payload': {
+                        'serialno': "123-456-678",
+                        'temparature': "XXX",
+                        'token': token, 
                     }
-                    await websocket.send(json.dumps(statusData))
-                    if (lockList[1]):
-                        break
-
-                    statusResponse = await websocket.recv()
-                    if lockList[1]:
-                        break
-
-                    statusResponseData = json.loads(statusResponse)
-                    print(f'send_websockrt_every10s')
-                    machineGetStatus = statusResponseData['status']
-                    machineGetType = statusResponseData['type']
-
-                    if lockList[1]:
-                        break
-
-                    if machineGetStatus == 1:
-                        lockList[0].acquire()
-                        if 'ads' in machineGetType:
+                }
+                await websocket.send(json.dumps(statusData))
+                if getThreadStatus() == THREAD_STOPPING:
+                    break
+
+                statusResponse = await websocket.recv()
+                if getThreadStatus() == THREAD_STOPPING:
+                    break
+
+                statusResponseData = json.loads(statusResponse)
+                print(f'send_websockrt_every10s')
+                machineGetStatus = statusResponseData['status']
+                machineGetType = statusResponseData['type']
+
+                if getThreadStatus() == THREAD_STOPPING:
+                    break
+
+                if machineGetStatus == 1:
+                    if 'ads' in machineGetType:
+                        getDBLock(DBLOCK_ADS).acquire()
+                        try:
                             send_get_ads_info()
-                        if 'machine' in machineGetType:
+                        except:
+                            pass
+                        getDBLock(DBLOCK_ADS).release()
+                    if 'machine' in machineGetType:
+                        getDBLock(DBLOCK_MACHINE).acquire()
+                        try:
                             send_get_machine_info()
-                        if 'product' in machineGetType:
+                        except:
+                            pass
+                        getDBLock(DBLOCK_MACHINE).release()
+                    if 'product' in machineGetType:
+                        getDBLock(DBLOCK_PRODUCT).acquire()
+                        try:
                             send_get_products_info()
-                        lockList[0].release()
-
-            else:
+                        except:
+                            pass
+                        getDBLock(DBLOCK_PRODUCT).release()
+
+        except (ConnectionClosed):
+            try:
+                websocket = await websockets.connect('wss://212.224.86.112:8443', ssl = ssl_context)
+            except:
                 pass
-
-            # if stopConnect:
-            #     print('websocket_close')
-            #     await websocket.close()
+
+    try:
+        websocket.close()
     except:
         pass
-        # global_ads = db.get_ad()
-        # global_produts = db.get_products()
-        # global_machines = db.get_machines()
-    lockList[1] = 2
+
+    setThreadStatus(THREAD_FINISHED)
 
 def close_connect():
-    pass
-# asyncio.get_event_loop().run_until_complete(connect_to_server())
+    pass

config/utils.py

@@ -1,9 +1,7 @@
 import os
+import threading
 # dbPath = './DbFuncs/sql.db'
 
-# screenX = 600
-# screenY = 900
-
 # itemLength = 220
 
 
@@ -20,11 +18,34 @@ def write_to_file(data, filename):
         file.write(data)
     print("Stored blob data into: ", filename, "\n")
 
-lockList = []
-def initLock(lock):
-    global lockList
-    lockList.append(lock)
-def initThreadLock(lock):
-    global lockList
-    lockList.append(lock)
+DBLOCK_ADS = 0
+DBLOCK_MACHINE = 1
+DBLOCK_PRODUCT = 2
+dbLockList = []
+def initDBLock():
+    global dbLockList
+    dbLockList.append(threading.Lock())
+    dbLockList.append(threading.Lock())
+    dbLockList.append(threading.Lock())
+
+def getDBLock(db):
+    return dbLockList[db]
+
+THREAD_INIT = 0
+THREAD_RUNNING = 1
+THREAD_STOPPING = 2
+THREAD_FINISHED = 3
+
+threadStatus = []
+def initThreadLock():
+    global threadStatus
+    threadStatus.append(THREAD_INIT)
+
+def setThreadStatus(status):
+    global threadStatus
+    threadStatus[0] = status
+
+def getThreadStatus():
+    global threadStatus
+    return threadStatus[0]