Upload New File

cam_two.py

+import cv2
+import json
+from socket import *
+from tool.utils import load_class_names, plot_boxes_cv2
+import queue, _thread, threading, time
+import numpy as np
+import subprocess
+
+# windows下obs推拉流 2拆分版本 测试成功
+
+ip_add = '192.168.1.118'
+# ip_add = '127.0.0.1'
+
+
+command = ['ffmpeg',
+           '-re',
+           '-i', '-',
+           '-r', '15',
+           '-c:v', 'libx264',
+           #    '-preset','superfast',
+           #    '-maxrate', '3000k',
+           #    '-bufsize', '6000k',
+           # '-pix_fmt', 'yuv420p',
+           # '-g', '50',
+           # '-s', '640x360',
+           # '-c:a', 'aac',
+           # '-b:a', '160k',
+           # '-ac', '2',
+           # '-ar', '44100',
+           # '-b:a', '160k',
+           '-f', 'flv',
+           'rtmp://demo.wangdongdong.xyz/live/skd_2']
+process = subprocess.Popen(command, shell=False, stdin=subprocess.PIPE)
+# print("subprocess established!")
+
+# ip_add = ''
+
+def send_from(arr, dest):
+    view = memoryview(arr).cast('B')
+    while len(view):
+        nsent = dest.send(view)
+        view = view[nsent:]
+
+
+def recv_into(arr, source):
+    view = memoryview(arr).cast('B')
+    while len(view):
+        nrecv = source.recv_into(view)
+        view = view[nrecv:]
+
+
+s = socket(AF_INET, SOCK_STREAM)
+s.bind(('', 25000))
+s.listen(1)
+c, a = s.accept()
+print("edge 1 is connected")
+c_2 = socket(AF_INET, SOCK_STREAM)
+c_2.connect((ip_add, 25002))
+print("edge 2 is connected")
+qsize = 10
+boxQue = queue.Queue(qsize)
+img_sent = queue.Queue(qsize * 10)
+lock = threading.Lock()
+# time.sleep(10)
+
+fps = 0
+fps_dis = 0
+
+
+def recv_box():
+    lth = np.zeros(shape=(1,), dtype=np.int64)
+    while 1:
+        if boxQue.full():
+            # print('box is full')
+            time.sleep(0.1)
+        else:
+            recv_into(lth, c_2)
+            if lth[0] == 0:
+                lock.acquire()
+                boxQue.put([0])
+                lock.release()
+                continue
+            arr = np.zeros(shape=(1, lth[0], 7), dtype=np.float32)
+            recv_into(arr, c_2)
+            box = arr.tolist()
+            for i in range(lth[0]):
+                box[0][i][-1] = np.int64(box[0][i][-1])
+            lock.acquire()
+            boxQue.put(box)
+            lock.release()
+
+
+# sum_flag = np.zeros(shape=(1,), dtype=np.int32)
+# def recv_flag():
+#     global sum_flag
+#     recv_into(sum_flag, c)
+#     print('done')
+
+# def cam_send():
+#     cap = cv2.VideoCapture(0)
+#     flag = cap.isOpened()
+#     print(flag)
+#     # _thread.start_new_thread(recv_flag, ())
+#     cnt_arr = np.zeros(shape=(1,), dtype=np.int32)
+#     while 1:
+#         _, img = cap.read()
+#         send_from(img, c)
+#         send_from(np.array([np.sum(img)]), c)
+#         recv_into(cnt_arr, c)
+#         if cnt_arr[0]>=5:
+#             break
+#     while 1:
+#         while img_sent.full():
+#             # print('sent is full')
+#             time.sleep(0.1)
+#         _, img = cap.read()
+#         # print(img)
+#         send_from(img, c)
+#         lock.acquire()
+#         img_sent.put(img)
+#         lock.release()
+#         # print(np.sum(img))
+#         if cv2.waitKey(1) & 0xFF == ord('q'):
+#             break
+#     cap.release()
+
+cap = cv2.VideoCapture('rtmp://demo.wangdongdong.xyz/live/skd')
+flag = cap.isOpened()
+# print("cap is opened: ", flag)
+print("任务生成成功，已成功获取视频流：", flag)
+print("开始多雾节点协同计算")
+# while(True):
+#     # 获取一帧
+#     ret, frame = cap.read()
+#     cv2.imshow('frame', frame)
+#     if cv2.waitKey(1) == ord('q'):
+#         break
+
+img_read = ''
+lock_img = threading.Lock()
+def cam_read():
+    global cap, img_read
+    # print("no connection in the stream, reconnecting")
+    time.sleep(1)
+    cnt_arr = np.zeros(shape=(1,), dtype=np.int32)
+    while True:
+        if not cap.isOpened():
+            # _, frame = None
+            print('error while subprocess not running')
+            cam_send()
+        break
+
+    while True:
+        _, img = cap.read()
+        img = cv2.resize(img, (640, 480))
+        lock_img.acquire()
+        img_read = img
+        lock_img.release()
+
+
+def cam_send():
+    # while True:
+    #     _, img = cap.read()
+    #     # print(type(img), img.shape[:])
+    #     # cv2.imshow("img", img)
+    #     # time.sleep(0.1)
+    #     send_from(img, c)
+    #     send_from(np.array([np.sum(img)]), c)
+    #     recv_into(cnt_arr, c)
+    #     print("cnt_arr<5,loop!")
+    #     if cnt_arr[0] >= 5:
+    #         break
+
+    global img_read
+    while True:
+        while img_sent.full() or img_read == '':
+            # print('sent is full')
+            time.sleep(0.1)
+        # _, img = cap.read()
+        lock_img.acquire()
+        img = img_read.copy()
+        img_read = ''
+        lock_img.release()
+        # img = cv2.resize(img, (640, 480))
+        # cv2.imshow("img", img)   #########
+        send_from(img, c)
+        lock.acquire()
+        img_sent.put(img)
+        lock.release()
+        if cv2.waitKey(1) == 27:
+            break
+        else:
+            # print('key is wrong')
+            pass
+
+            # boxes, confs, clss = trt_yolo.detect(frame, 0.3)
+            # frame = vis.draw_bboxes(frame, boxes, confs, clss)
+            # cv2.putText(frame, "FPS: " + str(round(fps, 2)), (10, 50), font, 3, (255, 50, 0), 3)
+            # toc = time.time()
+            # curr_fps = 1.0 / (toc - tic)
+            # fps = curr_fps if fps == 0.0 else (fps * 0.95 + curr_fps * 0.05)
+            # tic = toc
+            # computing_time = (1 / fps) * 1000
+            # cv2.putText(frame, str(round(computing_time, 2)) + " ms", (10, 90), font, 3, (255, 50, 0), 3)
+            # ret_toRTSP, frame_toRTSP = cv2.imencode('.png', frame)
+            # process.stdin.write(frame_toRTSP.tobytes())
+            # # process.stdin.write(frame.tostring())
+            # # cv2.imshow("cap",frame)
+
+
+def fps_update():
+    global fps, fps_dis
+    while 1:
+        time.sleep(10)
+        print(fps)
+        fps_dis = fps / 10
+        fps = 0
+
+
+_thread.start_new_thread(recv_box, ())
+_thread.start_new_thread(cam_read, ())
+_thread.start_new_thread(cam_send, ())
+_thread.start_new_thread(fps_update, ())
+
+# def get_box():
+#     while boxQue.empty():
+#         time.sleep(0.1)
+#     lock.acquire()
+#     box = boxQue.get()
+#     lock.release()
+#     return box
+
+
+namesfile = 'data/coco.names'
+class_names = load_class_names(namesfile)
+while (1):
+    # get a frame
+    while img_sent.empty() or boxQue.empty():
+        # print('sent or box are empty')
+        time.sleep(0.1)
+    # print("empty end")
+    lock.acquire()
+    img = img_sent.get()
+    boxes = boxQue.get()
+    lock.release()
+    # print(np.sum(img))
+    # start = time.time()
+    if boxes[0] == 0:
+        # print("img boxes is empty!")
+        pass
+    else:
+        img = plot_boxes_cv2(img, boxes[0], 'predictions.jpg', class_names)
+    img = cv2.putText(img, 'FPS: {}'.format(fps_dis), (100, 100), cv2.FONT_HERSHEY_SIMPLEX, 2, (0, 0, 0), 2)
+    # print(img)
+    cv2.imshow('fps:', img)
+    fps += 1
+    if cv2.waitKey(1) & 0xFF == ord('q'):
+        break
+    ##########################################################################################################
+    # ret_toRTSP, frame_toRTSP = cv2.imencode('.png', img)
+    # # print(frame_toRTSP)
+    # process.stdin.write(frame_toRTSP.tobytes())
+    ##########################################################################################################
+    # end = time.time()
+    # print('time: ', end - start)
+    # cv2.imshow('fps:', img)
+    # fps += 1
+    # if cv2.waitKey(1) & 0xFF == ord('q'):
+    #     break
+
+cv2.destroyAllWindows()
+c.close()
+s.close()