From ef8c2ecc9c0e679490157720ba88c8f1b92838be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A8=81=E5=8B=9D=20=E5=BC=B5?= Date: Mon, 29 Apr 2024 14:49:02 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=20function=20=E4=B8=8A?= =?UTF-8?q?=E5=82=B3=20=E5=8D=97=E5=81=B4=E9=96=80=E9=80=B2=E5=87=BA?= =?UTF-8?q?=E8=B3=87=E6=96=99-1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .idea/Stable_Python.iml | 6 + .../South_side/Class/API_class.py | 318 +----------------- .../South_side/test_1115_main.py | 51 ++- 3 files changed, 60 insertions(+), 315 deletions(-) diff --git a/.idea/Stable_Python.iml b/.idea/Stable_Python.iml index 86086e3..7c6e936 100644 --- a/.idea/Stable_Python.iml +++ b/.idea/Stable_Python.iml @@ -5,6 +5,9 @@ + + + + \ No newline at end of file diff --git a/Yuntech_NET_Stable_0429/South_side/Class/API_class.py b/Yuntech_NET_Stable_0429/South_side/Class/API_class.py index 0b86264..dcd7a53 100644 --- a/Yuntech_NET_Stable_0429/South_side/Class/API_class.py +++ b/Yuntech_NET_Stable_0429/South_side/Class/API_class.py @@ -50,10 +50,10 @@ class Parking_spaces_API(): return - -class Start_API(): +class South_API(): def __init__(self): self.token = self.User_token()['token'] + def User_token(self): API = "/Users/authenticate" data = { @@ -65,318 +65,13 @@ class Start_API(): data = response.json() return data - # 獲取CAM全部資料 - def get_all_cam_data(self): - API = "/api/Parking_spaces_cam" - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - data = Parking_spaces_API(API, headers).get() - return data - - - # 獲取ROI某筆資料 - def get_roi_pass_data(self,parking_spaces_name): - API = "/api/Parking_spaces_roi_pass/" + parking_spaces_name - headers = CaseInsensitiveDict(); - headers['Accept'] = 'application/json' - data = Parking_spaces_API(API, headers).get() - return data - - - # 獲取CAM某停車場資料 - def get_one_cam_data(self,parking_spaces_name): - API = "/api/Parking_spaces_cam/parking_spaces_name-" + parking_spaces_name - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - cam_data = Parking_spaces_API(API, headers).get() - return cam_data - - # 獲取ROI某筆資料 - def get_roi_violation_data(self, parking_spaces_name): - API = "/api/Parking_spaces_roi_violation/parking_spaces_name-" + parking_spaces_name - headers = CaseInsensitiveDict(); - headers['Accept'] = 'application/json' - data = Parking_spaces_API(API, headers).get() - return data - - - - -class Pass_API(): - def __init__(self): + # 更新 token + def token_update(self): self.token = self.User_token()['token'] - def User_token(self): - API = "/Users/authenticate" - data = { - 'username': "admin", - 'password': "admin" - } - headers = {'Content-type': 'application/json'} - response = Parking_spaces_API(API, headers).post(data) - data = response.json() - return data - - # 獲取get_cam的資料 - def get_cam(self,parking_spaces_name): - API = "/api/Parking_spaces_cam/parking_spaces_name-" + parking_spaces_name - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - - data = Parking_spaces_API(API, headers).get() - return data - - # 獲取get_cam_ptz的資料 - def get_cam_ptz(self,parking_spaces_name): - API = "/api/Parking_spaces_cam_ptz_pass/parking_spaces_name-" + parking_spaces_name - headers = CaseInsensitiveDict(); - headers['Accept'] = 'application/json' - data = Parking_spaces_API(API, headers).get() - return data - - # 更新cam_ptz的資料 - def put_cam_ptz(self,data, IP): - parking_spaces_name = data['parking_spaces_name'] - port = data['rtsp_port'] - API = "/api/Parking_spaces_cam_ptz_pass/parking_spaces_name-" + parking_spaces_name + "-IP-" + IP + "-port-" + port - headers = { - 'accept': '*/*', - 'Content-type': 'application/json'} - res = Parking_spaces_API(API, headers).put(data) - print(res) - # 上傳資料(當前) - def post_data_instant(self,data): - API = "/api/Parking_spaces_instant/" - headers = {'Content-type': 'application/json'} - response = Parking_spaces_API(API, headers).post(data) - print(response) - - - # 上傳資料(歷史) - def post_data_history(self,data): - API = "/api/Parking_spaces_history/" - headers = {'Content-type': 'application/json'} - response = Parking_spaces_API(API, headers).post(data) - print(response) - - - # 刪除資料 - def delete_data(self,plate): - API = "/api/Parking_spaces_instant/" + plate - headers = {'Content-type': 'application/json'} - response = Parking_spaces_API(API, headers).delete() - print(response) - - - - # 字幕機 數據+1 - def LED_add(self,parking_spaces_name): - API = "/api/Parking_spaces_total_table/Parking_spaces_total_table_add-"+parking_spaces_name - headers = CaseInsensitiveDict(); - headers['Accept'] = 'application/json' - data = Parking_spaces_API(API, headers).get() - - - - # 字幕機 數據-1 - def LED_minus(self,parking_spaces_name): - API = "/api/Parking_spaces_total_table/Parking_spaces_total_table_subtract-"+parking_spaces_name - headers = CaseInsensitiveDict(); - headers['Accept'] = 'application/json' - data = Parking_spaces_API(API, headers).get() - - - def get_num(self,parking_spaces_name): - API = "/api/Parking_spaces_total_table/" + parking_spaces_name - headers = CaseInsensitiveDict(); - headers['Accept'] = 'application/json' - data = Parking_spaces_API(API, headers).getSingleData() - num = data['parking_spaces_now_num'][0] - return num - - - -class Car_num_check_API(): - def __init__(self): - self.token = self.User_token()['token'] - def User_token(self): - API = "/Users/authenticate" - data = { - 'username': "admin", - 'password': "admin" - } - headers = {'Content-type': 'application/json'} - response = Parking_spaces_API(API, headers).post(data) - data = response.json() - return data - - # 獲取get_cam的資料 - def get_cam(self,parking_spaces_name): - API = "/api/Parking_spaces_cam/parking_spaces_name-" + parking_spaces_name - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - - data = Parking_spaces_API(API, headers).get() - return data - - # 獲取get_cam_ptz的資料 - def get_cam_ptz(self,parking_spaces_name): - API = "/api/Parking_spaces_cam_ptz_car_num_check/parking_spaces_name-" + parking_spaces_name - headers = CaseInsensitiveDict(); - headers['Accept'] = 'application/json' - data = Parking_spaces_API(API, headers).get() - return data - - # 更新cam_ptz的資料 - def put_can_ptz(self,data): - parking_spaces_name = data['parking_spaces_name'] - serial_num = data['serial_num'] - port = data['rtsp_port'] - API = "/api/Parking_spaces_cam_ptz_car_num_check/parking_spaces_name-" + parking_spaces_name + "-serial_num-" + serial_num - # print(API) - headers = { - 'accept': '*/*', - 'Content-type': 'application/json'} - res = Parking_spaces_API(API, headers).put(data) - # print(res) - - - # 獲取get_cam_roi的資料 - def get_cam_roi(self,parking_spaces_name): - API = "/api/Parking_spaces_roi_car_num_check/parking_spaces_name-" + parking_spaces_name - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - data = Parking_spaces_API(API, headers).get() - return data - - def get_num(self,parking_spaces_name): - API = "/api/Parking_spaces_total_table/" + parking_spaces_name - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - data = Parking_spaces_API(API, headers).getSingleData() - num = data['parking_spaces_now_num'][0] - return num - - def get_all_num(self,parking_spaces_name): - API = "/api/Parking_spaces_total_table/" + parking_spaces_name - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - data = Parking_spaces_API(API, headers).getSingleData() - num = data['parking_spaces_total_num'][0] - return num - - def put_parking_spaces_name_num(self,parking_spaces_name,num): - API = "/api/Parking_spaces_total_table/" + parking_spaces_name - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - data = Parking_spaces_API(API, headers).getSingleData() - data_1 = { - "parking_spaces_name":data['parking_spaces_name'][0], - "parking_spaces_total_num": data['parking_spaces_total_num'][0], - "longitude": data['longitude'][0], - "latitude": data['latitude'][0], - "parking_spaces_now_num": str(num), - - } - API = "/api/Parking_spaces_total_table/" + parking_spaces_name - # print(API) - headers = { - 'accept': '*/*', - 'Content-type': 'application/json'} - res = Parking_spaces_API(API, headers).put(data_1) - print(res) - - API = "/api/Parking_spaces_lcd_instand/parking_space_area-" + parking_spaces_name - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - data = Parking_spaces_API(API, headers).get() - for i in range(0,len(data)): - if data['parking_spaces_name'][i] == parking_spaces_name: - data_1 = { - "parking_spaces_name":data['parking_spaces_name'][i], - "lcd_ip":data['lcd_ip'][i], - "parking_space_amount":str(num) - } - API = "/api/Parking_spaces_lcd_instand/" + data['lcd_ip'][i] - # print(API) - headers = { - 'accept': '*/*', - 'Content-type': 'application/json'} - res = Parking_spaces_API(API, headers).put(data_1) - print(res) - - - - - #return num - -class Violation_API(): - def __init__(self): - self.token = self.User_token()['token'] - def User_token(self): - API = "/Users/authenticate" - data = { - 'username': "admin", - 'password': "admin" - } - headers = {'Content-type': 'application/json'} - response = Parking_spaces_API(API, headers).post(data) - data = response.json() - return data - - # 獲取get_cam的資料 - def get_cam(self,parking_spaces_name): - API = "/api/Parking_spaces_cam/parking_spaces_name-" + parking_spaces_name - headers = { - 'Content-type': 'application/json', - "Authorization": self.token - } - - data = Parking_spaces_API(API, headers).get() - return data - - # 獲取get_cam_ptz的資料 - def get_cam_ptz(self,parking_spaces_name): - API = "/api/Parking_spaces_cam_ptz_violation/parking_spaces_name-" + parking_spaces_name - headers = CaseInsensitiveDict(); - headers['Accept'] = 'application/json' - data = Parking_spaces_API(API, headers).get() - return data - - # 更新cam_ptz的資料 - def put_can_ptz(self,data): - parking_spaces_name = data['parking_spaces_name'] - port = data['rtsp_port'] - serial_num = data['serial_num'] - API = "/api/Parking_spaces_cam_ptz_violation/parking_spaces_name-" + parking_spaces_name + "-rtsp_port-" + port+"-serial_num-"+serial_num - print(API) - headers = { - 'accept': '*/*', - 'Content-type': 'application/json'} - res = Parking_spaces_API(API, headers).put(data) - print(res) - - def post_violation_car(self,data): - API = "/api/Violation_car_table/" + def post_data(self, data): + API = "/api/Yuntech_in_car_table" headers = { 'Content-type': 'application/json', "Authorization": self.token @@ -399,3 +94,4 @@ class Violation_API(): + diff --git a/Yuntech_NET_Stable_0429/South_side/test_1115_main.py b/Yuntech_NET_Stable_0429/South_side/test_1115_main.py index 5cd955c..5f9f98f 100644 --- a/Yuntech_NET_Stable_0429/South_side/test_1115_main.py +++ b/Yuntech_NET_Stable_0429/South_side/test_1115_main.py @@ -8,7 +8,7 @@ import re from flask import render_template from flask import Flask, request, jsonify from pyModbusTCP.client import ModbusClient - +from Class.API_class import South_API import datetime import shutil @@ -421,12 +421,16 @@ class Webcam(threading.Thread): time.sleep(1) + + + class Camera(mp.Process): def __init__(self, data, parent=None): super().__init__(parent) self.url = data['rtsp_url'] self.parking_spaces_name = data['rtsp_name'] self.IO_IP = data['IP'] + self.api = South_API() def run(self): # self.IO = IO_Webapi(self.IO_IP) @@ -493,7 +497,6 @@ class Camera(mp.Process): self.licence_check(cars_data) self.IO_control() # self.all_open() - self.state_check() img = self.img.copy() self.show_img(car_img) @@ -529,6 +532,8 @@ class Camera(mp.Process): licence_img = car_img[licence_box_location[1]:licence_box_location[1] + licence_box_location[3], licence_box_location[0]:licence_box_location[0] + licence_box_location[2]] self.flag_1 = True + if len(self.licence_list)==0: + self.car_img = img result = self.ocr.licence_detect(licence_img) if result != [[]]: @@ -564,8 +569,8 @@ class Camera(mp.Process): self.licence_fps = self.licence_fps + 1 if self.licence_fps > 3 and len(self.licence_list) > 2: self.licence_num = self.licence_vote() - if self.licence_num in self.white_licence_list: - self.open = True + #if self.licence_num in self.white_licence_list: + self.open = True self.flag_1 = False self.flag_2 = True @@ -596,6 +601,7 @@ class Camera(mp.Process): print("open") self.text = "open" self.open_time = time.time() + self.state_check() self.flag_2 = False self.IO.IO.out_1_clicked() if self.open == True and (close_time - self.open_time) > 1: @@ -613,6 +619,23 @@ class Camera(mp.Process): # 判斷狀態並上傳到API def state_check(self): + if self.open == True: + if self.flag_2 == True: + print(f'{self.licence_num}-進入') + cv2.imwrite(f'img/licence/{self.licence_num}.jpg', self.car_img) # 写入图像路径 + img_base64 = self.img_to_base64(self.car_img) + data = { + "location": "南側門", + "license_plate_number": self.licence_num, + "car_img": img_base64, + "in_time": "2024-02-04T19:15:09.4371245+08:00", + "out_time": "2024-02-04T19:15:09.4371245+08:00" + } + try: + self.api.post_data(data) + except BaseException as e: + print(e) + print("error_api") pass ## 將指定時間影片儲存 @@ -718,6 +741,18 @@ class Camera(mp.Process): cv2.imshow(f'{self.parking_spaces_name}', img) cv2.waitKey(1) + # 圖片 轉 base64 + def img_to_base64(self,img): + _, buffer = cv2.imencode('.jpg', img) + img_bytes = buffer.tobytes() + + # 編碼為 base64 字串 + img_base64 = base64.b64encode(img_bytes).decode('utf-8') + + # 拼接 base64 字串 + data_uri = "data:image/jpeg;base64," + img_base64 + + return data_uri if __name__ == '__main__': @@ -730,6 +765,14 @@ if __name__ == '__main__': # rtsp_url_list = [basepath+'test_2.avi']# , basepath+'test_1.avi'] rtsp_name_list = ['南側門出口', '南側門入口'] IO_IP_list = ["192.168.100.201", "192.168.100.204"] + + # 開發用 + # rtsp_url_list = ['rtsp://admin:admin123@192.168.100.102', 'rtsp://admin:admin123@192.168.100.101'] + rtsp_url_list = [basepath+'test1.avi']# , basepath+'test_1.avi'] + rtsp_name_list = ['南側門入口'] + IO_IP_list = [ "192.168.1.200"] + + webcam_list = [] for i in range(0, len(rtsp_url_list)): data = {