208 lines
6.4 KiB
Python
208 lines
6.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
红绿灯信号测试客户端
|
|
用于测试红绿灯TCP服务器的功能
|
|
"""
|
|
|
|
import socket
|
|
import json
|
|
import time
|
|
import threading
|
|
import random
|
|
|
|
class TrafficLightTestClient:
|
|
def __init__(self, host='localhost', port=8082):
|
|
self.host = host
|
|
self.port = port
|
|
self.socket = None
|
|
self.running = False
|
|
|
|
def connect(self):
|
|
"""连接到红绿灯TCP服务器"""
|
|
try:
|
|
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
self.socket.connect((self.host, self.port))
|
|
print(f"✅ 成功连接到红绿灯服务器 {self.host}:{self.port}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 连接失败: {e}")
|
|
return False
|
|
|
|
def disconnect(self):
|
|
"""断开连接"""
|
|
self.running = False
|
|
if self.socket:
|
|
try:
|
|
self.socket.close()
|
|
print("🔌 连接已断开")
|
|
except:
|
|
pass
|
|
|
|
def send_signal(self, device_id, ns_red=0, ns_yellow=0, ns_green=0,
|
|
ew_red=0, ew_yellow=0, ew_green=0):
|
|
"""发送红绿灯信号"""
|
|
signal = {
|
|
"device_id": device_id,
|
|
"DI-01": 0,
|
|
"DI-02": 0,
|
|
"DI-11": ns_red, # 北红
|
|
"DI-12": ns_yellow, # 北黄
|
|
"DI-13": ns_green, # 北绿
|
|
"DI-14": ew_red, # 东红
|
|
"DI-15": ew_yellow, # 东黄
|
|
"DI-16": ew_green, # 东绿
|
|
"DI-17": 0,
|
|
"DI-18": 0
|
|
}
|
|
|
|
try:
|
|
message = json.dumps(signal) + '\n'
|
|
self.socket.send(message.encode('utf-8'))
|
|
print(f"📡 发送信号: {device_id} - NS:{self._get_state(ns_red, ns_yellow, ns_green)} EW:{self._get_state(ew_red, ew_yellow, ew_green)}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ 发送信号失败: {e}")
|
|
return False
|
|
|
|
def _get_state(self, red, yellow, green):
|
|
"""获取信号状态描述"""
|
|
if red == 1:
|
|
return "红"
|
|
elif yellow == 1:
|
|
return "黄"
|
|
elif green == 1:
|
|
return "绿"
|
|
else:
|
|
return "未知"
|
|
|
|
def simulate_traffic_light_cycle(self, device_id, cycles=5):
|
|
"""模拟红绿灯周期"""
|
|
print(f"🚦 开始模拟设备 {device_id} 的红绿灯周期...")
|
|
|
|
# 红绿灯状态序列：南北绿-东西红 -> 南北黄-东西红 -> 南北红-东西绿 -> 南北红-东西黄
|
|
states = [
|
|
(0, 0, 1, 1, 0, 0), # NS绿, EW红
|
|
(0, 1, 0, 1, 0, 0), # NS黄, EW红
|
|
(1, 0, 0, 0, 0, 1), # NS红, EW绿
|
|
(1, 0, 0, 0, 1, 0), # NS红, EW黄
|
|
]
|
|
|
|
for cycle in range(cycles):
|
|
print(f"\n--- 第 {cycle + 1} 个周期 ---")
|
|
for i, (ns_r, ns_y, ns_g, ew_r, ew_y, ew_g) in enumerate(states):
|
|
if not self.running:
|
|
break
|
|
|
|
success = self.send_signal(device_id, ns_r, ns_y, ns_g, ew_r, ew_y, ew_g)
|
|
if not success:
|
|
break
|
|
|
|
# 每个状态持续3秒
|
|
time.sleep(3)
|
|
|
|
if not self.running:
|
|
break
|
|
|
|
def send_random_signals(self, device_id, duration=30):
|
|
"""发送随机信号"""
|
|
print(f"🎲 开始发送随机信号，持续 {duration} 秒...")
|
|
|
|
start_time = time.time()
|
|
while time.time() - start_time < duration and self.running:
|
|
# 随机生成信号状态
|
|
ns_state = random.choice([(1,0,0), (0,1,0), (0,0,1)]) # 红、黄、绿
|
|
ew_state = random.choice([(1,0,0), (0,1,0), (0,0,1)])
|
|
|
|
success = self.send_signal(device_id,
|
|
ns_state[0], ns_state[1], ns_state[2],
|
|
ew_state[0], ew_state[1], ew_state[2])
|
|
if not success:
|
|
break
|
|
|
|
# 随机间隔 1-3 秒
|
|
time.sleep(random.uniform(1, 3))
|
|
|
|
def test_single_device():
|
|
"""测试单个设备"""
|
|
client = TrafficLightTestClient()
|
|
|
|
if not client.connect():
|
|
return
|
|
|
|
client.running = True
|
|
|
|
try:
|
|
# 模拟正常的红绿灯周期
|
|
client.simulate_traffic_light_cycle("TL_001", cycles=3)
|
|
|
|
print("\n" + "="*50)
|
|
print("测试完成！")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n⏹️ 测试被用户中断")
|
|
finally:
|
|
client.disconnect()
|
|
|
|
def test_multiple_devices():
|
|
"""测试多个设备并发"""
|
|
devices = ["TL_001", "TL_002", "TL_003"]
|
|
clients = []
|
|
threads = []
|
|
|
|
print(f"🚀 启动多设备测试，设备数量: {len(devices)}")
|
|
|
|
def device_worker(device_id):
|
|
client = TrafficLightTestClient()
|
|
clients.append(client)
|
|
|
|
if client.connect():
|
|
client.running = True
|
|
client.send_random_signals(device_id, duration=20)
|
|
|
|
client.disconnect()
|
|
|
|
try:
|
|
# 启动多个线程模拟多个设备
|
|
for device_id in devices:
|
|
thread = threading.Thread(target=device_worker, args=(device_id,))
|
|
thread.start()
|
|
threads.append(thread)
|
|
time.sleep(1) # 错开连接时间
|
|
|
|
# 等待所有线程完成
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
print("\n" + "="*50)
|
|
print("多设备测试完成！")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n⏹️ 测试被用户中断")
|
|
# 停止所有客户端
|
|
for client in clients:
|
|
client.running = False
|
|
|
|
def main():
|
|
print("🚦 红绿灯信号测试客户端")
|
|
print("="*50)
|
|
|
|
while True:
|
|
print("\n请选择测试模式:")
|
|
print("1. 单设备测试（模拟正常红绿灯周期）")
|
|
print("2. 多设备测试（并发随机信号）")
|
|
print("3. 退出")
|
|
|
|
choice = input("\n请输入选择 (1-3): ").strip()
|
|
|
|
if choice == '1':
|
|
test_single_device()
|
|
elif choice == '2':
|
|
test_multiple_devices()
|
|
elif choice == '3':
|
|
print("👋 再见！")
|
|
break
|
|
else:
|
|
print("❌ 无效选择，请重新输入")
|
|
|
|
if __name__ == "__main__":
|
|
main() |