添加测试海康api的程序

This commit is contained in:
haotian 2025-08-26 17:43:13 +08:00
parent 106a7417ff
commit 0db2a85153
11 changed files with 94 additions and 2 deletions

View File

@ -0,0 +1,3 @@
from utils.haikang_util import HaikangUtil

Binary file not shown.

After

Width:  |  Height:  |  Size: 144 KiB

View File

@ -157,6 +157,7 @@ class HaiKangSettings:
HAIKANG_DOOR_STATES_URL = '/api/v1/door/states'
HAIKANG_DOOR_DOCONTROL_URL = '/api/acs/v1/door/doControl'
HAIKANG_DOOR_ENVENTS_URL = '/api/acs/v2/door/events'
HAIKANG_DOOR_SEARCH = '/api/resource/v2/door/search'
HAIKANG_APPLICATION_ONETOMANY_URL = '/api/frs/v1/application/oneToMany'
HAIKANG_PICTURE_CHECK_URL = '/api/frs/v1/face/picture/check'
HAIKANG_FACECAPATURE_SEARCH = '/api/frs/v1/event/face_capture/search'

View File

@ -0,0 +1,5 @@
from utils.haikang_util import HaikangUtil
import asyncio
access_token = asyncio.run(HaikangUtil.get_access_token())
print("access_token:", access_token)

View File

@ -0,0 +1,6 @@
from utils.haikang_util import HaikangUtil
import asyncio
result = asyncio.run(HaikangUtil.get_door_list_v2())
print(result)

View File

@ -0,0 +1,8 @@
from utils.haikang_util import HaikangUtil
import asyncio
door_index_codes = ['D01']
data = asyncio.run( HaikangUtil.get_door_status(door_index_codes))
print("门禁状态", data)

View File

@ -0,0 +1,8 @@
from utils.haikang_util import HaikangUtil
import asyncio
door_index_code = ['D01']
control_type = 2 # 0-常开, 1-门闭, 2-门开, 3-常闭
result = asyncio.run(HaikangUtil.door_do_control(door_index_code, control_type))
print("测试控制门禁", result)

View File

@ -0,0 +1,7 @@
from utils.haikang_util import HaikangUtil
import asyncio
door_index_code = ['D01']
result = asyncio.run( HaikangUtil.query_door_events_v2(door_index_code))
print("查询门禁点事件", result)

View File

@ -0,0 +1,8 @@
from utils.haikang_util import HaikangUtil
import asyncio
result = asyncio.run( HaikangUtil.get_face_group())
print("测试查询人脸分组", result)

View File

@ -0,0 +1,17 @@
import base64
from utils.haikang_util import HaikangUtil
import asyncio
with open("75c03e462769c81b6a8513d90ff2a27d.jpg", "rb") as image_file:
encoded_bytes = base64.b64encode(image_file.read())
# 解码为字符串默认为bytes方便打印或传输
encoded_str = encoded_bytes.decode('utf-8')
# print(encoded_str)
faceGroupIndexCodes = []
result = asyncio.run( HaikangUtil.face_group_1vN_search(encoded_str,
faceGroupIndexCodes=faceGroupIndexCodes))
print("测试人脸1vN检索", result)

View File

@ -125,6 +125,35 @@ class HaikangUtil:
return [False, back]
# 查询门禁点列表v2
@classmethod
async def get_door_list_v2(cls, pageNo: int = 1, pageSize: int = 10):
"""获取门禁点列表v2
Args:
pageNo (int, optional): 页码. Defaults to 1.
pageSize (int, optional): 每页个数. Defaults to 10.
Returns:
_type_: _description_
"""
url = f'{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_SEARCH}'
body_dict ={
"pageNo": max(pageNo, 1),
"pageSize": min(pageSize, 999),
}
body_json = json.dumps(body_dict, separators=(",", ":"))
headers = cls.build_signed_headers('POST', url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK)
back = await cls.send_request('POST', url, headers, body_json)
if back["code"] == "0":
logger.info("查询门禁点列表成功")
return [True, back["data"]]
else:
logger.error("查询门禁点列表失败 ", back["code"], back["msg"])
return [False, back["code"], back["msg"]]
# 查询门禁状态
@classmethod
async def get_door_status(cls, door_index_codes:list):
@ -200,8 +229,8 @@ class HaikangUtil:
# three_months_ago_iso = three_months_ago.isoformat()
body_dict ={
"pageNo": min(args.get("pageNo", 1), 1),
"pageSize": max(args.get("pageSize", 10), 999),
"pageNo": max(args.get("pageNo", 1), 1),
"pageSize": min(args.get("pageSize", 10), 999),
"doorIndexCode": door_index_code,
# 排序字段