From 0ecd4a0ec8c4c5585cbd8975d7786c5618814381 Mon Sep 17 00:00:00 2001
From: Junjie <DELL@qq.com>
Date: 星期三, 03 十二月 2025 08:28:59 +0800
Subject: [PATCH] #
---
camera_api.py | 323 +++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 323 insertions(+), 0 deletions(-)
diff --git a/camera_api.py b/camera_api.py
new file mode 100644
index 0000000..27f783c
--- /dev/null
+++ b/camera_api.py
@@ -0,0 +1,323 @@
+from fastapi import FastAPI, HTTPException, Query
+from fastapi.responses import JSONResponse, FileResponse
+from pydantic import BaseModel
+from typing import Optional, List
+import os
+import uvicorn
+from camera_manager import HikCameraManager
+
+# 鍒涘缓FastAPI搴旂敤
+app = FastAPI(
+ title="娴峰悍鐩告満API鏈嶅姟",
+ description="閫氳繃HTTP鎺ュ彛鎺у埗娴峰悍宸ヤ笟鐩告満鎷嶇収",
+ version="1.0.0"
+)
+
+# 鍏ㄥ眬鐩告満绠＄悊鍣�
+camera_manager = HikCameraManager()
+# 瀛樺偍鍦板潃
+storageAddress = "D:\\pic\\"
+
+class CaptureRequest(BaseModel):
+ """鎷嶇収璇锋眰鍙傛暟"""
+ ip: str
+ filename: str
+ save_bmp: bool = True
+ save_jpg: bool = True
+ timeout: int = 10000
+
+
+class CaptureResponse(BaseModel):
+ """鎷嶇収鍝嶅簲"""
+ success: bool
+ message: str
+ files: List[str] = []
+ camera_info: dict = {}
+
+
+@app.on_event("startup")
+async def startup_event():
+ """鍚姩鏃跺垵濮嬪寲SDK"""
+ try:
+ camera_manager.initialize()
+ print("鐩告満API鏈嶅姟鍚姩鎴愬姛")
+ except Exception as e:
+ print(f"鍒濆鍖栧け璐�: {e}")
+
+
+@app.on_event("shutdown")
+async def shutdown_event():
+ """鍏抽棴鏃舵竻鐞嗚祫婧�"""
+ try:
+ camera_manager.finalize()
+ print("鐩告満API鏈嶅姟宸插叧闂�")
+ except Exception as e:
+ print(f"娓呯悊璧勬簮澶辫触: {e}")
+
+
+@app.get("/")
+async def root():
+ """鏍硅矾寰�"""
+ return {
+ "service": "娴峰悍鐩告満API鏈嶅姟",
+ "version": "1.0.0",
+ "endpoints": {
+ "GET /cameras": "鍒楀嚭鎵�鏈夌浉鏈�",
+ "POST /capture": "鎷嶇収锛圝SON body锛�",
+ "GET /capture": "鎷嶇収锛圲RL鍙傛暟锛�",
+ "GET /image/{filename}": "涓嬭浇鍥剧墖"
+ }
+ }
+
+
+@app.get("/cameras")
+async def list_cameras():
+ """
+ 鍒楀嚭鎵�鏈夊彲鐢ㄧ浉鏈�
+
+ 杩斿洖:
+ 鐩告満鍒楄〃鍙婅缁嗕俊鎭�
+ """
+ try:
+ cameras = camera_manager.list_cameras()
+
+ # 鏍煎紡鍖栬繑鍥炴暟鎹�
+ camera_list = []
+ for cam in cameras:
+ camera_list.append({
+ "index": cam['index'],
+ "type": cam['type'],
+ "model": cam['model'],
+ "serial": cam['serial'],
+ "ip": cam['ip'] if cam['ip'] else None,
+ "user_defined_name": cam['user_defined_name'] if cam['user_defined_name'] else None
+ })
+
+ return {
+ "success": True,
+ "count": len(camera_list),
+ "cameras": camera_list
+ }
+
+ except Exception as e:
+ raise HTTPException(status_code=500, detail=f"鑾峰彇鐩告満鍒楄〃澶辫触: {str(e)}")
+
+
+@app.post("/capture")
+async def capture_image_post(request: CaptureRequest):
+ """
+ 鎷嶇収鎺ュ彛锛圥OST鏂瑰紡锛�
+
+ 鍙傛暟:
+ - ip: 鐩告満IP鍦板潃
+ - filename: 淇濆瓨鐨勬枃浠跺悕锛堜笉鍚墿灞曞悕锛�
+ - save_bmp: 鏄惁淇濆瓨BMP鏍煎紡锛堥粯璁rue锛�
+ - save_jpg: 鏄惁淇濆瓨JPG鏍煎紡锛堥粯璁rue锛�
+ - timeout: 瓒呮椂鏃堕棿锛堟绉掞紝榛樿3000锛�
+
+ 杩斿洖:
+ 鎷嶇収缁撴灉鍜屼繚瀛樼殑鏂囦欢鍒楄〃
+ """
+ try:
+ # 鎵撳紑鐩告満
+ camera_info = camera_manager.open_camera(ip=request.ip)
+
+ # 鎷嶇収
+ saved_files = camera_manager.capture_image(
+ filename=request.filename,
+ storageAddress=storageAddress,
+ save_bmp=request.save_bmp,
+ save_jpg=request.save_jpg,
+ timeout=request.timeout
+ )
+
+ # 鍏抽棴鐩告満
+ camera_manager.close_camera()
+
+ return CaptureResponse(
+ success=True,
+ message="鎷嶇収鎴愬姛",
+ files=saved_files,
+ camera_info={
+ "type": camera_info['type'],
+ "model": camera_info['model'],
+ "serial": camera_info['serial'],
+ "ip": camera_info['ip']
+ }
+ )
+
+ except Exception as e:
+ # 纭繚鍏抽棴鐩告満
+ try:
+ camera_manager.close_camera()
+ except:
+ pass
+
+ raise HTTPException(status_code=500, detail=f"鎷嶇収澶辫触: {str(e)}")
+
+
+@app.get("/capture")
+async def capture_image_get(
+ ip: str = Query(..., description="鐩告満IP鍦板潃"),
+ filename: str = Query(..., description="淇濆瓨鐨勬枃浠跺悕锛堜笉鍚墿灞曞悕锛�"),
+ save_bmp: bool = Query(True, description="鏄惁淇濆瓨BMP鏍煎紡"),
+ save_jpg: bool = Query(True, description="鏄惁淇濆瓨JPG鏍煎紡"),
+ timeout: int = Query(3000, description="瓒呮椂鏃堕棿锛堟绉掞級")
+):
+ """
+ 鎷嶇収鎺ュ彛锛圙ET鏂瑰紡锛�
+
+ 鍙傛暟:
+ - ip: 鐩告満IP鍦板潃
+ - filename: 淇濆瓨鐨勬枃浠跺悕锛堜笉鍚墿灞曞悕锛�
+ - save_bmp: 鏄惁淇濆瓨BMP鏍煎紡锛堥粯璁rue锛�
+ - save_jpg: 鏄惁淇濆瓨JPG鏍煎紡锛堥粯璁rue锛�
+ - timeout: 瓒呮椂鏃堕棿锛堟绉掞紝榛樿3000锛�
+
+ 绀轰緥:
+ GET /capture?ip=192.168.1.100&filename=test_image
+ """
+ request = CaptureRequest(
+ ip=ip,
+ filename=filename,
+ save_bmp=save_bmp,
+ save_jpg=save_jpg,
+ timeout=timeout
+ )
+ return await capture_image_post(request)
+
+
+@app.get("/capture_by_serial")
+async def capture_by_serial(
+ serial: str = Query(..., description="鐩告満搴忓垪鍙�"),
+ filename: str = Query(..., description="淇濆瓨鐨勬枃浠跺悕锛堜笉鍚墿灞曞悕锛�"),
+ save_bmp: bool = Query(True, description="鏄惁淇濆瓨BMP鏍煎紡"),
+ save_jpg: bool = Query(True, description="鏄惁淇濆瓨JPG鏍煎紡"),
+ timeout: int = Query(3000, description="瓒呮椂鏃堕棿锛堟绉掞級")
+):
+ """
+ 閫氳繃搴忓垪鍙锋媿鐓�
+
+ 绀轰緥:
+ GET /capture_by_serial?serial=00J12345678&filename=test_image
+ """
+ try:
+ camera_info = camera_manager.open_camera(serial=serial)
+
+ saved_files = camera_manager.capture_image(
+ filename=filename,
+ save_bmp=save_bmp,
+ save_jpg=save_jpg,
+ timeout=timeout
+ )
+
+ camera_manager.close_camera()
+
+ return CaptureResponse(
+ success=True,
+ message="鎷嶇収鎴愬姛",
+ files=saved_files,
+ camera_info={
+ "type": camera_info['type'],
+ "model": camera_info['model'],
+ "serial": camera_info['serial'],
+ "ip": camera_info.get('ip', '')
+ }
+ )
+
+ except Exception as e:
+ try:
+ camera_manager.close_camera()
+ except:
+ pass
+ raise HTTPException(status_code=500, detail=f"鎷嶇収澶辫触: {str(e)}")
+
+
+@app.get("/capture_by_index")
+async def capture_by_index(
+ index: int = Query(0, description="鐩告満绱㈠紩"),
+ filename: str = Query(..., description="淇濆瓨鐨勬枃浠跺悕锛堜笉鍚墿灞曞悕锛�"),
+ save_bmp: bool = Query(True, description="鏄惁淇濆瓨BMP鏍煎紡"),
+ save_jpg: bool = Query(True, description="鏄惁淇濆瓨JPG鏍煎紡"),
+ timeout: int = Query(3000, description="瓒呮椂鏃堕棿锛堟绉掞級")
+):
+ """
+ 閫氳繃绱㈠紩鎷嶇収
+
+ 绀轰緥:
+ GET /capture_by_index?index=0&filename=test_image
+ """
+ try:
+ camera_info = camera_manager.open_camera(index=index)
+
+ saved_files = camera_manager.capture_image(
+ filename=filename,
+ storageAddress=storageAddress,
+ save_bmp=save_bmp,
+ save_jpg=save_jpg,
+ timeout=timeout
+ )
+
+ camera_manager.close_camera()
+
+ return CaptureResponse(
+ success=True,
+ message="鎷嶇収鎴愬姛",
+ files=saved_files,
+ camera_info={
+ "type": camera_info['type'],
+ "model": camera_info['model'],
+ "serial": camera_info['serial'],
+ "ip": camera_info.get('ip', '')
+ }
+ )
+
+ except Exception as e:
+ try:
+ camera_manager.close_camera()
+ except:
+ pass
+ raise HTTPException(status_code=500, detail=f"鎷嶇収澶辫触: {str(e)}")
+
+
+@app.get("/image/{filename}")
+async def get_image(filename: str):
+ """
+ 涓嬭浇鍥剧墖
+
+ 鍙傛暟:
+ filename: 鏂囦欢鍚嶏紙鍚墿灞曞悕锛�
+
+ 绀轰緥:
+ GET /image/test_image.jpg
+ """
+
+ filepath = storageAddress + filename
+ if not os.path.exists(filepath):
+ raise HTTPException(status_code=404, detail="鏂囦欢涓嶅瓨鍦�")
+
+ return FileResponse(
+ filepath,
+ media_type="application/octet-stream",
+ filename=filename
+ )
+
+
+@app.get("/health")
+async def health_check():
+ """鍋ュ悍妫�鏌�"""
+ return {
+ "status": "healthy",
+ "service": "camera-api",
+ "sdk_initialized": camera_manager.is_initialized
+ }
+
+
+if __name__ == "__main__":
+ # 鍚姩鏈嶅姟
+ uvicorn.run(
+ app,
+ host="0.0.0.0",
+ port=8000,
+ log_level="info"
+ )
--
Gitblit v1.9.1