200 lines
7.9 KiB
Python
200 lines
7.9 KiB
Python
# hardware/transmitter.py
|
||
import socket
|
||
import threading
|
||
from config.ini_manager import ini_manager
|
||
from config.settings import app_set_config
|
||
import time
|
||
|
||
class TransmitterController:
|
||
def __init__(self):
|
||
self.upper_ip = ini_manager.upper_transmitter_ip
|
||
self.upper_port = ini_manager.upper_transmitter_port
|
||
self.lower_ip = ini_manager.lower_transmitter_ip
|
||
self.lower_port = ini_manager.lower_transmitter_port
|
||
|
||
# 存储最新重量值
|
||
self.latest_weights = {1: None, 2: None}
|
||
# 存储连接状态
|
||
self.connection_status = {1: False, 2: False}
|
||
# 线程控制
|
||
self.running = True
|
||
self.threads = {}
|
||
# 连接配置
|
||
self.TIMEOUT = 5 # 连接超时时间
|
||
self.BUFFER_SIZE = 1024
|
||
|
||
# 启动后台接收线程
|
||
self._start_receiver_threads()
|
||
|
||
def _start_receiver_threads(self):
|
||
"""启动后台接收线程"""
|
||
for transmitter_id in [1, 2]:
|
||
if (transmitter_id == 1 and self.upper_ip and self.upper_port) or \
|
||
(transmitter_id == 2 and self.lower_ip and self.lower_port):
|
||
thread = threading.Thread(
|
||
target=self._continuous_receiver,
|
||
args=(transmitter_id,),
|
||
daemon=True,
|
||
name=f'transmitter_receiver_{transmitter_id}'
|
||
)
|
||
thread.start()
|
||
self.threads[transmitter_id] = thread
|
||
print(f"启动变送器 {transmitter_id} 后台接收线程")
|
||
|
||
def _continuous_receiver(self, transmitter_id):
|
||
"""后台持续接收数据的线程函数"""
|
||
while self.running:
|
||
IP = None
|
||
PORT = None
|
||
|
||
if transmitter_id == 1:
|
||
IP = self.upper_ip
|
||
PORT = self.upper_port
|
||
elif transmitter_id == 2:
|
||
IP = self.lower_ip
|
||
PORT = self.lower_port
|
||
|
||
if not IP or not PORT:
|
||
time.sleep(5)
|
||
continue
|
||
|
||
sock = None
|
||
try:
|
||
# 创建连接
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(self.TIMEOUT)
|
||
sock.connect((IP, PORT))
|
||
self.connection_status[transmitter_id] = True
|
||
print(f"变送器 {transmitter_id} 连接成功: {IP}:{PORT}")
|
||
|
||
# 持续接收数据
|
||
while self.running:
|
||
try:
|
||
data = sock.recv(self.BUFFER_SIZE)
|
||
if data:
|
||
# 提取有效数据包
|
||
packet = self.get_latest_valid_packet(data)
|
||
if packet:
|
||
# 解析重量
|
||
weight = self.parse_weight(packet)
|
||
if weight is not None:
|
||
self.latest_weights[transmitter_id] = weight
|
||
# 可选：打印接收到的重量
|
||
# print(f"变送器 {transmitter_id} 重量: {weight}")
|
||
else:
|
||
# 连接关闭
|
||
print(f"变送器 {transmitter_id} 连接关闭")
|
||
break
|
||
except socket.timeout:
|
||
# 超时是正常的，继续接收
|
||
continue
|
||
except Exception as e:
|
||
print(f"接收数据异常: {e}")
|
||
break
|
||
|
||
except ConnectionRefusedError:
|
||
print(f"变送器 {transmitter_id} 连接失败：{IP}:{PORT} 拒绝连接")
|
||
except Exception as e:
|
||
print(f"变送器 {transmitter_id} 异常：{e}")
|
||
finally:
|
||
self.connection_status[transmitter_id] = False
|
||
if sock:
|
||
try:
|
||
sock.close()
|
||
except:
|
||
pass
|
||
# 重试间隔
|
||
time.sleep(3)
|
||
|
||
# 直接读取 变送器返回的数据（从缓存中获取）
|
||
def read_data_sub(self, transmitter_id):
|
||
|
||
"""
|
||
Args: transmitter_id 为1 表示上料斗, 为2 表示下料斗
|
||
return: 读取成功返回重量 weight: int, 失败返回 None
|
||
"""
|
||
# 直接返回缓存的最新重量值
|
||
return self.latest_weights.get(transmitter_id)
|
||
|
||
def get_connection_status(self, transmitter_id):
|
||
"""
|
||
获取变送器连接状态
|
||
Args: transmitter_id 为1 表示上料斗, 为2 表示下料斗
|
||
return: 连接状态 bool
|
||
"""
|
||
return self.connection_status.get(transmitter_id, False)
|
||
|
||
def stop(self):
|
||
"""停止后台线程"""
|
||
self.running = False
|
||
# 等待线程结束
|
||
for thread in self.threads.values():
|
||
if thread.is_alive():
|
||
thread.join(timeout=2)
|
||
print("变送器后台接收线程已停止")
|
||
|
||
def get_latest_valid_packet(self, raw_data):
|
||
"""
|
||
解决TCP粘包:
|
||
从原始数据中，筛选所有有效包，返回最新的一个有效包
|
||
有效包标准: 1. 能UTF-8解码 2. 按逗号拆分≥3个字段 3. 第三个字段含数字（重量）
|
||
"""
|
||
DELIMITER = b'\r\n'
|
||
# 1. 按分隔符拆分，过滤空包
|
||
packets = [p for p in raw_data.split(DELIMITER) if p]
|
||
if not packets:
|
||
return None
|
||
|
||
valid_packets = []
|
||
for packet in packets:
|
||
try:
|
||
# 过滤无效ASCII字符（只保留可见字符）
|
||
valid_chars = [c for c in packet if 32 <= c <= 126]
|
||
filtered_packet = bytes(valid_chars)
|
||
# 2. 验证解码
|
||
data_str = filtered_packet.decode('utf-8').strip()
|
||
# 3. 验证字段数量
|
||
parts = data_str.split(',')
|
||
if len(parts) < 3:
|
||
continue
|
||
# 4. 验证重量字段含数字
|
||
weight_part = parts[2].strip()
|
||
if not any(char.isdigit() for char in weight_part):
|
||
continue
|
||
# 满足所有条件，加入有效包列表
|
||
valid_packets.append(packet)
|
||
except (UnicodeDecodeError, IndexError):
|
||
# 解码失败或字段异常，跳过该包
|
||
continue
|
||
|
||
# 返回最后一个有效包（最新），无有效包则返回None
|
||
return valid_packets[-1] if valid_packets else None
|
||
|
||
def parse_weight(self, packet_data):
|
||
"""解析重量函数：提取重量数值（如从 b'ST,NT,+0000175\r\n' 中提取 175)"""
|
||
try:
|
||
data_str = packet_data.decode('utf-8').strip()
|
||
parts = data_str.split(',')
|
||
# 确保有完整的数据包，三个字段
|
||
if len(parts) < 3:
|
||
print(f"parse_weight: 包格式错误（字段不足）：{data_str}")
|
||
return None
|
||
|
||
weight_part = parts[2].strip()
|
||
return int(''.join(filter(str.isdigit, weight_part)))
|
||
except (IndexError, ValueError, UnicodeDecodeError) as e:
|
||
# print(f"数据解析失败：{e}，原始数据包：{packet_data}")
|
||
return None
|
||
|
||
def read_data(self,transmitter_id):
|
||
"""获取重量函数：根据变送器ID获取当前重量,三次"""
|
||
max_try_times=5
|
||
try_times=0
|
||
while try_times<max_try_times:
|
||
weight=self.read_data_sub(transmitter_id)
|
||
if weight is not None:
|
||
return weight
|
||
try_times+=1
|
||
print(f'-----获取重量异常-------------- transmitter_id: {transmitter_id}')
|
||
print(f'-----获取重量异常-------------- transmitter_id: {transmitter_id}')
|
||
return None |