211 lines
7.4 KiB
Python
211 lines
7.4 KiB
Python
import argparse
|
|
import sys
|
|
import time
|
|
import configparser
|
|
|
|
from capture_thermal import save_img
|
|
from lxml import etree
|
|
import pandas as pd
|
|
import requests
|
|
|
|
|
|
def read_env_file(file_path):
|
|
# 创建一个 ConfigParser 对象
|
|
config = configparser.ConfigParser()
|
|
|
|
# 读取 .env 文件
|
|
with open(file_path, 'r') as f:
|
|
# 将 .env 文件内容转换为 configparser 可以解析的格式
|
|
config.read_string(f"[dummy_section]\n" + f.read())
|
|
|
|
# 获取参数
|
|
db_version = config.get('dummy_section', 'DB_VERSION')
|
|
db_cluster_name = config.get('dummy_section', 'DB_CLUSTER_NAME')
|
|
db_host = config.get('dummy_section', 'DB_HOST')
|
|
db_port = config.get('dummy_section', 'DB_PORT')
|
|
db_db_name = config.get('dummy_section', 'DB_DB_NAME')
|
|
db_uname = config.get('dummy_section', 'DB_UNAME')
|
|
db_passwd = config.get('dummy_section', 'DB_PASSWD')
|
|
db_table = config.get('dummy_section', 'DB_TABLE')
|
|
ipc_user = config.get('dummy_section', 'IPC_USER')
|
|
ipc_passwd = config.get('dummy_section', 'IPC_PASSWD')
|
|
|
|
return {
|
|
'DB_VERSION': db_version,
|
|
'DB_CLUSTER_NAME': db_cluster_name,
|
|
'DB_HOST': db_host,
|
|
'DB_PORT': db_port,
|
|
'DB_DB_NAME': db_db_name,
|
|
'DB_UNAME': db_uname,
|
|
'DB_PASSWD': db_passwd,
|
|
'DB_TABLE': db_table,
|
|
'IPC_USER': ipc_user,
|
|
'IPC_PASSWD': ipc_passwd
|
|
}
|
|
|
|
|
|
def get_ptz(url):
|
|
print("Feching PTZ status...")
|
|
response = requests.get(url)
|
|
print(f"Response Status Code: {response.status_code}")
|
|
# print(f"Response Content: {response.text}")
|
|
|
|
return response.text
|
|
|
|
|
|
def set_ptz(url, xml_str):
|
|
headers = {'Content-Type': 'application/xml'}
|
|
response = requests.put(url, data=xml_str, headers=headers)
|
|
print(f"Response Status Code: {response.status_code}")
|
|
# print(f"Response Content: {response.text}")
|
|
|
|
|
|
def send_xml_via_put_request(file_path, id_value, column_index, url):
|
|
df = pd.read_excel(file_path, sheet_name='Sheet2')
|
|
row = df[df['id'] == int(id_value-100)]
|
|
|
|
if row.empty:
|
|
print(f"No row found with id: {id_value}")
|
|
return
|
|
|
|
try:
|
|
xml_content = row.iloc[0, column_index+1]
|
|
# print(xml_content)
|
|
except IndexError:
|
|
print(f"Column index {column_index} is out of range.")
|
|
return
|
|
|
|
set_ptz(url, xml_content)
|
|
|
|
|
|
def find_not_nan(file_path, id_value):
|
|
lst_ret = []
|
|
df = pd.read_excel(file_path, sheet_name='Sheet2')
|
|
|
|
row = df[df['id'] == int(id_value-100)]
|
|
|
|
if row.empty:
|
|
print(f"No row found with id: {id_value}")
|
|
return
|
|
|
|
for i in range(0, 4):
|
|
row_idx = row.index[0]
|
|
is_cell_empty = df.iat[row_idx, i+1] != df.iat[row_idx, i+1]
|
|
if not is_cell_empty:
|
|
lst_ret.append(i)
|
|
|
|
return lst_ret
|
|
|
|
|
|
def valid_move(move):
|
|
ivalue = int(move)
|
|
if ivalue < -1800 or ivalue > 1800:
|
|
raise argparse.ArgumentTypeError(f"down and right must be within -1800 - 1800 (got {ivalue})")
|
|
|
|
return ivalue
|
|
|
|
|
|
def valid_id(id):
|
|
ivalue = int(id)
|
|
if ivalue < 101 or ivalue > 120:
|
|
raise argparse.ArgumentTypeError(f"ID must be an integer between 1 and 20 (got {ivalue})")
|
|
|
|
return ivalue
|
|
|
|
|
|
def modify_xml_ptz(xml_str, down=0, right=0, zoom_in=0):
|
|
root = etree.fromstring(xml_str.encode('utf-8'), parser=etree.XMLParser(remove_blank_text=True))
|
|
ns_old = 'http://www.isapi.org/ver20/XMLSchema'
|
|
ns_new = 'http://www.std-cgi.com/ver20/XMLSchema'
|
|
namespace = {'ns': ns_new}
|
|
|
|
elevation_element = root.find('.//ns:elevation', namespace)
|
|
if elevation_element is None:
|
|
namespace = {'ns': ns_old}
|
|
elevation_element = root.find('.//ns:elevation', namespace)
|
|
if elevation_element is None:
|
|
raise ValueError("elevation element not found in XML")
|
|
new_elevation = int(elevation_element.text) + down
|
|
new_elevation = new_elevation if new_elevation < 3600 else new_elevation - 3600
|
|
new_elevation = new_elevation if new_elevation > 0 else new_elevation + 3600
|
|
elevation_element.text = str(new_elevation)
|
|
|
|
azimuth_element = root.find('.//ns:azimuth', namespace)
|
|
if azimuth_element is None:
|
|
namespace = {'ns': ns_old}
|
|
azimuth_element = root.find('.//ns:azimuth', namespace)
|
|
if azimuth_element is None:
|
|
raise ValueError("azimuth element not found in XML")
|
|
new_azimuth = int(azimuth_element.text) + right
|
|
new_azimuth = new_azimuth if new_azimuth < 3600 else new_azimuth - 3600
|
|
new_elevation = new_elevation if new_elevation > 0 else new_elevation + 3600
|
|
azimuth_element.text = str(new_azimuth)
|
|
|
|
absolute_zoom_element = root.find('.//ns:absoluteZoom', namespace)
|
|
if absolute_zoom_element is None:
|
|
namespace = {'ns': ns_old}
|
|
absolute_zoom_element = root.find('.//ns:absoluteZoom', namespace)
|
|
if absolute_zoom_element is None:
|
|
raise ValueError("absoluteZoom element not found in XML")
|
|
new_zoom = int(absolute_zoom_element.text) + zoom_in
|
|
if new_zoom < 10:
|
|
print("zoom adjusted to minimum value 10")
|
|
new_zoom = 10
|
|
absolute_zoom_element.text = str(new_zoom)
|
|
|
|
xml_send = etree.tostring(root, pretty_print=True, encoding='utf-8').decode('utf-8').replace('PTZStatus', 'PTZData')
|
|
# xml_send = f'<?xml version="1.0" encoding="UTF-8"?>\n{xml_send}'
|
|
print(xml_send)
|
|
|
|
return xml_send
|
|
|
|
|
|
def argparser():
|
|
parser = argparse.ArgumentParser(description='Adjust camera settings')
|
|
parser.add_argument('--sleep_s', '-s', type=int, default=8, help='sleep time in seconds (default: 8)')
|
|
parser.add_argument('id', type=valid_id, help='camera id')
|
|
parser.add_argument('--file_path', '-f', type=str, help='path to the excel file')
|
|
parser.add_argument('--down', '-p', type=valid_move, default=0, help='move down')
|
|
parser.add_argument('--right', '-t', type=valid_move, default=0, help='move right')
|
|
parser.add_argument('--zoom_in', '-z', type=int, default=0, help='zoom in')
|
|
|
|
return parser.parse_args()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
args = argparser()
|
|
env_vars = read_env_file('./.env')
|
|
ipc_user = env_vars["IPC_USER"]
|
|
ipc_passwd = env_vars["IPC_PASSWD"]
|
|
get_url = f"http://{ipc_user}:{ipc_passwd}@192.168.137.{args.id}:80/ISAPI/PTZCtrl/channels/1/status"
|
|
put_url = f"http://{ipc_user}:{ipc_passwd}@192.168.137.{args.id}:80/ISAPI/PTZCtrl/channels/1/absolute"
|
|
|
|
if args.file_path:
|
|
lst_ret = find_not_nan(args.file_path, args.id)
|
|
print(lst_ret)
|
|
for i in lst_ret:
|
|
if 0 == i:
|
|
continue
|
|
send_xml_via_put_request(args.file_path, args.id, i, put_url)
|
|
print("Adjusting to ", i)
|
|
time.sleep(args.sleep_s)
|
|
print("sleep: ", args.sleep_s)
|
|
save_img(str(args.id), str(i), ipc_user, ipc_passwd)
|
|
|
|
if 0 in lst_ret:
|
|
send_xml_via_put_request(args.file_path, args.id, 0, put_url)
|
|
print("Adjusting to ", 0)
|
|
time.sleep(args.sleep_s)
|
|
print("sleep: ", args.sleep_s)
|
|
save_img(str(args.id), '0', ipc_user, ipc_passwd)
|
|
else:
|
|
xml_got = get_ptz(get_url)
|
|
print(xml_got)
|
|
if args.down != 0 or args.right != 0 or args.zoom_in != 0:
|
|
print(f"Adjusting to pan {args.down} tile {args.right} zoom {args.zoom_in}")
|
|
xml_send = modify_xml_ptz(xml_got, args.down, args.right, args.zoom_in)
|
|
set_ptz(put_url, xml_send)
|
|
print("sleep: ", args.sleep_s)
|
|
time.sleep(args.sleep_s)
|
|
save_img(str(args.id), '0', ipc_user, ipc_passwd)
|