FaulknerWu/smart-inspection
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

查验

Browse Source
This commit is contained in:
ywb
2026-06-16 14:17:05 +08:00
parent 62292edc70
commit 916b44bc3c
10 changed files with 725 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agv_app/utils/mission_executor.py
+80 -18
View File
@@ -28,10 +28,10 @@ from utils.nav2_navigator import Nav2Navigator, Nav2Status
logger = logging.getLogger(__name__)
ROS2_SETUP_CMD = "source /opt/ros/humble/setup.bash && source ~/agv_pro_ros2/install/setup.bash"
from config import ARM_CAMERA_CONFIG
from config import ARM_CAMERA_CONFIG, UPLOAD_CONFIG, ZHIJIAN_AUTH_TOKEN
ARM_CAMERA_SNAPSHOT = ARM_CAMERA_CONFIG["snapshot_url"]
PHOTOS_DIR = "/home/elephant/photos"
UPLOAD_URL = "https://ts.zhijian168.com/prod-api/file/uploadImage"
# UPLOAD_CONFIG["url"] 随环境切换动态变化，每次使用时直接读取
# 二维码扫描重试参数
QR_SCAN_TIMEOUT = 5 # 单次扫描超时
@@ -719,6 +719,7 @@ class MissionExecutorV3:
try:
resp = requests.get(ARM_CAMERA_SNAPSHOT, timeout=QR_SCAN_TIMEOUT)
if resp.status_code != 200 or not resp.content:
self._log(f" 📷 arm snapshot attempt {attempt+1}: HTTP {resp.status_code}, size={len(resp.content) if resp.content else 0}")
continue
arr = np.frombuffer(resp.content, dtype=np.uint8)
@@ -735,13 +736,18 @@ class MissionExecutorV3:
time.sleep(0.5)
return None
def _request_manual_qr(self) -> Optional[str]:
"""暂停任务，等待手动输入（支持重新扫描）"""
def _request_manual_qr(self, message: str = None) -> Optional[str]:
"""暂停任务，等待手动输入（支持重新扫描）
message: 自定义弹窗消息（None 则使用默认消息）"""
while True:
self.status = MissionStatus.WAITING_QR
self.report["status"] = "waiting_qr"
self.report["step"] = "等待手动输入二维码"
self._log(" ⌨️ 弹窗等待手动输入二维码...")
self.report["step"] = message or "等待手动输入二维码"
self.report["qr_message"] = message or "所有姿态均未识别到二维码，请手动输入："
if message:
self._log(f" ⌨️ {message}")
else:
self._log(" ⌨️ 弹窗等待手动输入二维码...")
self._qr_event.clear()
self._qr_event.wait() # 无限等待，直到 set_manual_qr 或 stop() 触发
@@ -773,8 +779,61 @@ class MissionExecutorV3:
# ==================== 机型查询 ====================
def _lookup_model(self, qr_value: Optional[str]) -> str:
"""TODO: 后续通过 HTTP 接口查询机型"""
return "机器1"
"""通过 /api/customs/printer 接口查询机型，同时更新查验计数
如果机型不在当前报关单中，弹窗要求重新扫码/输入"""
if not qr_value:
return "机器1"
while True:
if self._stop.is_set():
return "机器1"
try:
printer_url = f"http://127.0.0.1:5000/api/customs/printer?serialNumber={qr_value}"
self._log(f" 🔍 查询机型 → {printer_url}")
resp = requests.get(printer_url, timeout=10)
self._log(f" 📡 printer 响应 HTTP {resp.status_code}: {resp.text[:500]}")
if resp.status_code == 200:
data = resp.json()
if data.get("ok"):
model = data.get("modelName", "机器1")
inv_code = data.get("inventoryCode", "")
matched = data.get("matchedItem")
has_inspection = data.get("hasInspection", False)
self._log(f" 📊 解析结果: modelName={model}, inventoryCode={inv_code}, hasInspection={has_inspection}, matched={'yes' if matched else 'no'}")
if matched:
self._log(f" 🏷️ 机型: {model} (物料:{inv_code}) — 查验 {matched['inspected']}/{matched['quantify']}")
return model
elif has_inspection and not matched:
# 有查验但机型不在报关单中 → 弹窗
self._log(f" ⚠️ 机型「{model}」(物料:{inv_code})不在当前报关单中")
new_qr = self._request_manual_qr(
f"二维码「{qr_value}」对应机型「{model}」不在当前报关单中，\n请重新扫描或手动输入正确的二维码："
)
if new_qr is None or self._stop.is_set():
return model # 用户跳过/停止，保留原机型
qr_value = new_qr
continue # 用新二维码重试
else:
# 无查验或匹配成功
self._log(f" 🏷️ 机型: {model} (物料:{inv_code})")
return model
else:
self._log(f" ⚠️ printer 返回 ok=false: {data}")
# API 失败也弹窗
self._log(f" ⚠️ 查询机型失败, HTTP {resp.status_code}")
new_qr = self._request_manual_qr(
f"无法查询二维码「{qr_value}」对应的机型，\n请重新扫描或手动输入正确的二维码："
)
if new_qr is None or self._stop.is_set():
return "机器1"
qr_value = new_qr
except Exception as e:
self._log(f" ⚠️ 查询机型失败: {e}")
new_qr = self._request_manual_qr(
f"查询机型接口异常，请重新扫描或手动输入正确的二维码："
)
if new_qr is None or self._stop.is_set():
return "机器1"
qr_value = new_qr
@staticmethod
def _find_model(models: list, name: str) -> Optional[dict]:
@@ -852,7 +911,7 @@ class MissionExecutorV3:
try:
resp = requests.get(ARM_CAMERA_SNAPSHOT, timeout=10)
if resp.status_code != 200 or not resp.content:
logger.error("arm snapshot 请求失败")
self._log(f" 📷 拍照 arm snapshot 失败: HTTP {resp.status_code}, size={len(resp.content) if resp.content else 0}")
return None
# 生成文件名（用于上传）
@@ -861,13 +920,14 @@ class MissionExecutorV3:
# 直接上传到服务器（不保存本地）
if qr_value:
self._log(f" 📷 拍照成功 {len(resp.content)} bytes → {fname}")
self._upload_photo_bytes(fname, resp.content, qr_value, upload_index)
else:
self._log(" ⚠️ 无二维码，跳过上传")
return fname # 返回文件名（用于日志）
except Exception as e:
logger.error(f"拍照异常: {e}")
self._log(f" ❌ 拍照异常: {e}")
return None
def _upload_photo(self, filepath: str, serial_number: str, index: int) -> bool:
@@ -880,13 +940,14 @@ class MissionExecutorV3:
"""
try:
filename = os.path.basename(filepath)
headers = {
"Authorization": "Bearer eyJhbGciOiJIUzUxMiJ9.eyJ1c2VyX2lkIjoxLCJ1c2VyX2tleSI6ImZhNTNkZTZiLWE3NjYtNDZmNC05MDUyLTQ2MjUzZTAyNjdmNSIsInVzZXJuYW1lIjoiYWRtaW4ifQ.lC4vKThZo4aAOLsekm2kPgaEJRqRx-YDQWKfHFqxdPNESCKy57l3eIqaKTj2ZjAMaoYAwYlMrv5M1zAOJsO_PA"
}
headers = {"Authorization": ZHIJIAN_AUTH_TOKEN}
upload_url = UPLOAD_CONFIG["url"]
self._log(f" 📤 上传请求 → {upload_url} | serialNumber={serial_number} | index={index} | file={filename}")
with open(filepath, "rb") as f:
files = {"file": (filename, f, "image/jpeg")}
data = {"serialNumber": serial_number, "index": str(index)}
resp = requests.post(UPLOAD_URL, files=files, data=data, headers=headers, timeout=30)
resp = requests.post(upload_url, files=files, data=data, headers=headers, timeout=30)
self._log(f" 📡 上传响应 HTTP {resp.status_code}: {resp.text[:300]}")
if resp.status_code == 200:
self._log(f" ☁️ 上传成功 [{index}]: {filename}")
return True
@@ -907,12 +968,13 @@ class MissionExecutorV3:
index: 上传序号（从1开始递增）
"""
try:
headers = {
"Authorization": "Bearer eyJhbGciOiJIUzUxMiJ9.eyJ1c2VyX2lkIjoxLCJ1c2VyX2tleSI6ImZhNTNkZTZiLWE3NjYtNDZmNC05MDUyLTQ2MjUzZTAyNjdmNSIsInVzZXJuYW1lIjoiYWRtaW4ifQ.lC4vKThZo4aAOLsekm2kPgaEJRqRx-YDQWKfHFqxdPNESCKy57l3eIqaKTj2ZjAMaoYAwYlMrv5M1zAOJsO_PA"
}
headers = {"Authorization": ZHIJIAN_AUTH_TOKEN}
upload_url = UPLOAD_CONFIG["url"]
self._log(f" 📤 上传请求(内存) → {upload_url} | serialNumber={serial_number} | index={index} | file={filename}")
files = {"file": (filename, image_data, "image/jpeg")}
data = {"serialNumber": serial_number, "index": str(index)}
resp = requests.post(UPLOAD_URL, files=files, data=data, headers=headers, timeout=30)
resp = requests.post(upload_url, files=files, data=data, headers=headers, timeout=30)
self._log(f" 📡 上传响应 HTTP {resp.status_code}: {resp.text[:300]}")
if resp.status_code == 200:
self._log(f" ☁️ 上传成功 [{index}]: {filename}")
return True