329 lines
11 KiB
Python
329 lines
11 KiB
Python
import socket
|
||
from rest_framework.exceptions import ParseError
|
||
import json
|
||
import re
|
||
import time
|
||
from django.core.cache import cache
|
||
from apps.utils.thread import MyThread
|
||
import uuid
|
||
import logging
|
||
import threading
|
||
import requests
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
|
||
myLogger = logging.getLogger('log')
|
||
|
||
def get_checksum(body_msg):
|
||
return sum(body_msg) & 0xFF
|
||
|
||
def handle_bytes(arr):
|
||
if len(arr) < 8:
|
||
return f"返回数据长度错误-{arr}"
|
||
|
||
if arr[0] != 0xEB or arr[1] != 0x90:
|
||
return "数据头不正确"
|
||
|
||
|
||
# 读取长度信息
|
||
length_arr = arr[2:4][::-1] # 反转字节
|
||
length = int.from_bytes(length_arr, byteorder='little', signed=True) # 小端格式
|
||
|
||
# 提取内容
|
||
body = arr[4:4 + length - 3]
|
||
|
||
# 校验和检查
|
||
check_sum = get_checksum(body)
|
||
if check_sum != arr[length + 1]:
|
||
return "校验错误"
|
||
|
||
|
||
# 尾部标识检查
|
||
if arr[length + 2] != 0xFF or arr[length + 3] != 0xFE:
|
||
return "尾错误"
|
||
|
||
content = body.decode('utf-8')
|
||
|
||
res = json.loads(content)
|
||
|
||
return res[0]
|
||
|
||
def get_tyy_data_t(host, port, tid):
|
||
cd_thread_key_id = f"cd_thread_{host}_{port}_id"
|
||
cd_thread_key_val = f"cd_thread_{host}_{port}_val"
|
||
sc = None
|
||
def connect_and_send(retry=1):
|
||
nonlocal sc
|
||
try:
|
||
if sc is None:
|
||
sc = socket.socket()
|
||
sc.connect((host, int(port)))
|
||
sc.sendall(b"R")
|
||
except BrokenPipeError:
|
||
if retry > 0:
|
||
connect_and_send(retry-1)
|
||
else:
|
||
if sc:
|
||
try:
|
||
sc.close()
|
||
except Exception:
|
||
pass
|
||
sc = None
|
||
except OSError as e:
|
||
sc = None
|
||
cache.set(cd_thread_key_val, {"err_msg": f"采集器连接失败-{str(e)}"})
|
||
except ConnectionResetError:
|
||
sc = None
|
||
cache.set(cd_thread_key_val, {"err_msg": "采集器重置了连接"})
|
||
except socket.timeout:
|
||
sc = None
|
||
cache.set(cd_thread_key_val, {"err_msg": "采集器连接超时"})
|
||
except Exception as e:
|
||
sc = None
|
||
cache.set(cd_thread_key_val, {"err_msg": f"采集器连接失败-{str(e)}"})
|
||
|
||
while cache.get(cd_thread_key_id) == tid:
|
||
if cache.get(cd_thread_key_val) == "get":
|
||
cache.set(cd_thread_key_val, "working")
|
||
connect_and_send()
|
||
if sc is None:
|
||
continue
|
||
resp = sc.recv(1024)
|
||
res = handle_bytes(resp)
|
||
if isinstance(res, str):
|
||
cache.set(cd_thread_key_val, {"err_msg": f'采集器返回数据错误-{res}'})
|
||
elif not res:
|
||
cache.set(cd_thread_key_val, {"err_msg": f"采集器返回数据为空-{str(res)}"})
|
||
else:
|
||
myLogger.info(f"采集器返回数据-{res}")
|
||
cache.set(cd_thread_key_val, res)
|
||
time.sleep(0.3)
|
||
|
||
if sc:
|
||
try:
|
||
sc.close()
|
||
except Exception:
|
||
pass
|
||
|
||
def get_tyy_data_2(*args, retry=1):
|
||
host, port = args[0], int(args[1])
|
||
cd_thread_key_id = f"cd_thread_{host}_{port}_id"
|
||
cd_thread_key_val = f"cd_thread_{host}_{port}_val"
|
||
cd_thread_val_id = cache.get(cd_thread_key_id, default=None)
|
||
if cd_thread_val_id is None:
|
||
tid = uuid.uuid4()
|
||
cache.set(cd_thread_key_id, tid, timeout=10800)
|
||
cd_thread = MyThread(target=get_tyy_data_t, args=(host, port, tid), daemon=True)
|
||
cd_thread.start()
|
||
cache.set(cd_thread_key_val, "get")
|
||
num = 0
|
||
get_val = False
|
||
|
||
while True:
|
||
num += 1
|
||
if num > 8:
|
||
break
|
||
val = cache.get(cd_thread_key_val)
|
||
if isinstance(val, dict):
|
||
get_val = True
|
||
if "err_msg" in val:
|
||
raise ParseError(val["err_msg"])
|
||
return val
|
||
time.sleep(0.3)
|
||
|
||
if not get_val and retry > 0:
|
||
cache.set(cd_thread_key_id, None)
|
||
get_tyy_data_2(*args, retry=retry-1)
|
||
|
||
|
||
sc_all = {}
|
||
sc_lock = threading.Lock()
|
||
|
||
def get_tyy_data_1(*args, retry=1):
|
||
host, port = args[0], int(args[1])
|
||
global sc_all
|
||
sc = None
|
||
|
||
def connect_and_send(retry=1):
|
||
nonlocal sc
|
||
sc = sc_all.get(f"{host}_{port}", None)
|
||
try:
|
||
if sc is None:
|
||
sc = socket.socket()
|
||
sc.settimeout(5) # 设置超时
|
||
sc.connect((host, port))
|
||
sc_all[f"{host}_{port}"] = sc
|
||
else:
|
||
# 清空接收缓冲区
|
||
sc.settimeout(0.1) # 设置短暂超时
|
||
for _ in range(5):
|
||
try:
|
||
data = sc.recv(65536)
|
||
if not data:
|
||
break
|
||
except (socket.timeout, BlockingIOError):
|
||
break
|
||
sc.settimeout(5) # 恢复原超时设置
|
||
sc.sendall(b"R")
|
||
except BrokenPipeError:
|
||
if retry > 0:
|
||
if sc:
|
||
try:
|
||
sc.close()
|
||
except Exception:
|
||
pass
|
||
sc_all.pop(f"{host}_{port}", None)
|
||
return connect_and_send(retry-1)
|
||
else:
|
||
if sc:
|
||
try:
|
||
sc.close()
|
||
except Exception:
|
||
pass
|
||
sc_all.pop(f"{host}_{port}", None)
|
||
sc = None
|
||
raise ParseError("采集器连接失败-管道重置")
|
||
except OSError as e:
|
||
if sc:
|
||
try:
|
||
sc.close()
|
||
except Exception:
|
||
pass
|
||
sc_all.pop(f"{host}_{port}", None)
|
||
sc = None
|
||
raise ParseError(f"采集器连接失败-{str(e)}")
|
||
except TimeoutError as e:
|
||
if sc:
|
||
try:
|
||
sc.close()
|
||
except Exception:
|
||
pass
|
||
sc_all.pop(f"{host}_{port}", None)
|
||
sc = None
|
||
raise ParseError(f"采集器连接超时-{str(e)}")
|
||
|
||
with sc_lock:
|
||
connect_and_send()
|
||
resp = sc.recv(1024)
|
||
res = handle_bytes(resp)
|
||
# myLogger.error(res)
|
||
if isinstance(res, str):
|
||
raise ParseError(f'采集器返回数据错误-{res}')
|
||
else:
|
||
return res
|
||
|
||
|
||
def get_tyy_data_3(*args, retry=2):
|
||
host, port = args[0], int(args[1])
|
||
for attempt in range(retry):
|
||
try:
|
||
# 每次请求都新建连接（确保无共享状态）
|
||
with socket.create_connection((host, port), timeout=10) as sc:
|
||
sc.sendall(b"R")
|
||
|
||
# 接收完整响应（避免数据不完整）
|
||
# resp = b""
|
||
# while True:
|
||
# chunk = sc.recv(4096)
|
||
# if not chunk:
|
||
# break
|
||
# resp += chunk
|
||
resp = sc.recv(4096)
|
||
if not resp:
|
||
raise ParseError("设备未启动")
|
||
|
||
res = handle_bytes(resp)
|
||
if isinstance(res, str):
|
||
raise ParseError(f"采集器返回数据错误: {res}")
|
||
return res
|
||
except (socket.timeout, ConnectionError) as e:
|
||
if attempt == retry - 1: # 最后一次尝试失败才报错
|
||
raise ParseError(f"采集器连接失败: {str(e)}")
|
||
time.sleep(0.5) # 失败后等待 1s 再重试
|
||
except ParseError:
|
||
raise
|
||
except Exception as e:
|
||
raise ParseError(f"未知错误: {str(e)}")
|
||
|
||
|
||
def _parse_vtfp_line(line):
|
||
"""解析 SVS 的 VTFP 应答行，返回 (识别字符, 字段列表)。
|
||
|
||
返回格式形如 ``VTFP 0 <用户自定义数据...>``：
|
||
- 第 1 个字段固定为 ``VTFP``
|
||
- 第 2 个字段为状态码（0 成功，非 0 失败，失败时后面无其他数据）
|
||
- 其余字段为用户在程序中自定义输出的数据（即识别出的字符）
|
||
|
||
分隔符按协议同时兼容空格和逗号。
|
||
"""
|
||
parts = [p for p in re.split(r"[,\s]+", line.strip()) if p != ""]
|
||
if not parts or parts[0].upper() != "VTFP":
|
||
raise ParseError(f"SVS返回格式错误-{line}")
|
||
if len(parts) < 2:
|
||
raise ParseError(f"SVS返回缺少状态码-{line}")
|
||
status = parts[1]
|
||
if status != "0":
|
||
raise ParseError(f"SVS触发失败,状态码-{status}")
|
||
fields = parts[2:] # 用户自定义数据（识别字符），可能含可选的流程单元号回显
|
||
char = " ".join(fields)
|
||
return char, fields
|
||
|
||
|
||
def get_svs_char(host, port, flow_unit=0, timeout=10):
|
||
"""连接 SVS（维视智造 VisionBank），用 VTFP 触发拍照并取回识别字符。
|
||
|
||
返回 ``{"char": 识别字符, "fields": [...], "raw": 原始应答行}``。
|
||
"""
|
||
cmd = f"VTFP {flow_unit}\r\n".encode("ascii")
|
||
try:
|
||
with socket.create_connection((host, int(port)), timeout=timeout) as sc:
|
||
sc.sendall(cmd)
|
||
buf = bytearray()
|
||
while True:
|
||
chunk = sc.recv(1024)
|
||
if not chunk:
|
||
break
|
||
buf.extend(chunk)
|
||
text = buf.decode("utf-8", errors="replace")
|
||
# VTFP 应答前可能先收到 VGIF 等行，定位到 VTFP 那一行再解析
|
||
for ln in text.split("\r\n"):
|
||
if ln.startswith("VTFP"):
|
||
char, fields = _parse_vtfp_line(ln)
|
||
return {"char": char, "fields": fields, "raw": ln}
|
||
except ParseError:
|
||
raise
|
||
except (socket.timeout, ConnectionError, OSError) as e:
|
||
raise ParseError(f"SVS连接失败-{str(e)}")
|
||
raise ParseError("SVS未返回有效数据")
|
||
|
||
|
||
def get_tyy_data(host, port, svs_host=None, svs_port=None, svs_flow=0):
|
||
"""获取采集器数据；若传入 svs_host/svs_port，并行获取 SVS 识别字符并合并返回。
|
||
|
||
采集器与 SVS 任一失败均整体抛错。
|
||
"""
|
||
port = int(port)
|
||
|
||
def fetch_tyy():
|
||
r = requests.get(f"http://127.0.0.1:2300?host={host}&port={port}")
|
||
res = r.json()
|
||
if "err_msg" in res:
|
||
raise ParseError(res["err_msg"])
|
||
return res
|
||
|
||
if not (svs_host and svs_port):
|
||
return fetch_tyy()
|
||
|
||
# 采集器与 SVS 并行触发，尽量贴近同一时刻；任一失败在 result() 处抛出
|
||
with ThreadPoolExecutor(max_workers=2) as ex:
|
||
tyy_future = ex.submit(fetch_tyy)
|
||
svs_future = ex.submit(get_svs_char, svs_host, svs_port, svs_flow)
|
||
res = tyy_future.result()
|
||
svs = svs_future.result()
|
||
|
||
if not isinstance(res, dict):
|
||
# 采集器返回的不是 dict 时，包成一个完整 dict
|
||
res = {"tyy": res}
|
||
res["svs_char"] = svs["char"]
|
||
res["svs_fields"] = svs["fields"]
|
||
res["svs_raw"] = svs["raw"]
|
||
return res |