From 2fa19599467263dcf582bb12906e03328e03b4a4 Mon Sep 17 00:00:00 2001
From: zhang <zc857179121@qq.com>
Date: 星期三, 02 七月 2025 13:12:26 +0800
Subject: [PATCH] 初版提交
---
common/api_client.py | 124 +++++++++++++++++++++++++++++++++++++++++
1 files changed, 124 insertions(+), 0 deletions(-)
diff --git a/common/api_client.py b/common/api_client.py
new file mode 100644
index 0000000..ffc818e
--- /dev/null
+++ b/common/api_client.py
@@ -0,0 +1,124 @@
+"""
+涓嶳CS绯荤粺閫氫俊
+"""
+import requests
+import json
+import logging
+import time
+from typing import Dict, Optional, Any
+from .data_models import (
+ APIResponse, create_error_response, ResponseCode
+)
+
+try:
+ from config.settings import (
+ RCS_SERVER_HOST, RCS_SERVER_PORT,
+ REQUEST_TIMEOUT, AGV_STATUS_API_ENDPOINT
+ )
+except ImportError:
+ RCS_SERVER_HOST = "10.10.10.156"
+ RCS_SERVER_PORT = 8088
+ REQUEST_TIMEOUT = 30
+ AGV_STATUS_API_ENDPOINT = "/api/open/algorithm/getAgv"
+ logging.warning("鏃犳硶浠巆onfig.settings瀵煎叆閰嶇疆锛屼娇鐢ㄩ粯璁ゅ��")
+
+
+class APIClient:
+ """HTTP API瀹㈡埛绔熀绫�"""
+
+ def __init__(self, base_url: str, timeout: int = 30):
+ """鍒濆鍖朅PI瀹㈡埛绔�"""
+ self.base_url = base_url.rstrip('/')
+ self.timeout = timeout
+ self.logger = logging.getLogger(__name__)
+ self.session = requests.Session()
+
+ self.session.headers.update({
+ 'Content-Type': 'application/json',
+ 'Accept': 'application/json'
+ })
+
+ def _make_request(self, method: str, endpoint: str, data: Any = None,
+ params: Dict = None) -> APIResponse:
+ """鍙戦�丠TTP璇锋眰"""
+ url = f"{self.base_url}{endpoint}"
+
+ try:
+ start_time = time.time()
+
+ request_kwargs = {
+ 'timeout': self.timeout,
+ 'params': params
+ }
+
+ if data is not None:
+ if isinstance(data, (dict, list)):
+ request_kwargs['json'] = data
+ else:
+ request_kwargs['data'] = json.dumps(data)
+
+ response = self.session.request(method, url, **request_kwargs)
+
+ end_time = time.time()
+ duration = (end_time - start_time) * 1000 # 杞崲涓烘绉�
+
+ self.logger.info(f"{method} {url} - {response.status_code} - {duration:.2f}ms")
+
+ try:
+ response_data = response.json()
+ except ValueError:
+ response_data = {"text": response.text}
+
+ if response.status_code == 200:
+ if isinstance(response_data, dict) and 'code' in response_data:
+ return APIResponse(
+ code=response_data.get('code', ResponseCode.SUCCESS),
+ msg=response_data.get('msg', '鎿嶄綔鎴愬姛'),
+ data=response_data.get('data')
+ )
+ else:
+ return APIResponse(
+ code=ResponseCode.SUCCESS,
+ msg='鎿嶄綔鎴愬姛',
+ data=response_data
+ )
+ else:
+ return create_error_response(
+ ResponseCode.SERVER_ERROR,
+ f"HTTP {response.status_code}: {response.text}"
+ )
+
+ except requests.exceptions.Timeout:
+ self.logger.error(f"璇锋眰瓒呮椂: {url}")
+ return create_error_response(ResponseCode.SERVER_ERROR, "璇锋眰瓒呮椂")
+
+ except requests.exceptions.ConnectionError:
+ self.logger.error(f"杩炴帴閿欒: {url}")
+ return create_error_response(ResponseCode.SERVER_ERROR, "杩炴帴閿欒")
+
+ except Exception as e:
+ self.logger.error(f"璇锋眰寮傚父: {url} - {str(e)}")
+ return create_error_response(ResponseCode.SERVER_ERROR, f"璇锋眰寮傚父: {str(e)}")
+
+
+class RCSAPIClient(APIClient):
+
+ def __init__(self, rcs_host: str = None, rcs_port: int = None, timeout: int = None):
+ """鍒濆鍖朢CS API瀹㈡埛绔�"""
+ rcs_host = rcs_host or RCS_SERVER_HOST
+ rcs_port = rcs_port or RCS_SERVER_PORT
+ timeout = timeout or REQUEST_TIMEOUT
+
+ base_url = f"http://{rcs_host}:{rcs_port}"
+ super().__init__(base_url, timeout)
+
+ def get_agv_status(self, agv_id: Optional[str] = None,
+ map_id: Optional[str] = None) -> APIResponse:
+ """鑾峰彇AGV鐘舵��"""
+ data = {}
+ if agv_id:
+ data['agvId'] = agv_id
+ if map_id:
+ data['mapId'] = map_id
+
+ return self._make_request('POST', AGV_STATUS_API_ENDPOINT, data=data)
\ No newline at end of file
--
Gitblit v1.9.1