zj/UPC-Resent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

自定义映射

Browse Source
This commit is contained in:
zj
2026-03-23 15:57:58 +08:00
parent 3ecba44f9e
commit e97d9c427a
8 changed files with 365 additions and 338 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
scripts/generate.py
View File

@@ -60,10 +60,11 @@ def get_template_path(protocol, template_type):
raise ValueError(f"未知的模板类型: {template_type}")
def generate_listener(device, global_config, mappings):
def generate_listener(device, global_config, mappings, command_sets, global_commands):
"""生成监听服务文件"""
protocol = device.get('protocol', 'tcp').lower()
template_file = get_template_path(protocol, 'listener')
listen_protocol = device.get('listen_protocol', 'tcp').lower()
device_protocol = device.get('device_protocol', 'tcp').lower()
template_file = get_template_path(listen_protocol, 'listener')
# 检查模板文件是否存在
if not os.path.exists(template_file):
@@ -73,16 +74,35 @@ def generate_listener(device, global_config, mappings):
with open(template_file, 'r', encoding='utf-8') as f:
template = f.read()
# 获取该设备的指令集
command_set_id = device.get('command_set')
if command_set_id and command_set_id in command_sets:
device_commands = command_sets[command_set_id].get('commands', {})
else:
# 兼容旧配置
device_commands = global_commands
# 构建命令映射字典字符串
# 1. 首先添加该设备指令集中所有指令的默认映射（指令名映射到自身）
device_mappings = {}
for cmd in device_commands.keys():
device_mappings[cmd] = cmd
# 2. 然后添加全局 mappings 中的映射（外部命令 -> 内部指令）
# 但只添加目标指令在该设备指令集中存在的映射
for external_cmd, internal_cmd in mappings.items():
if internal_cmd in device_commands:
device_mappings[external_cmd] = internal_cmd
# 3. 生成映射表字符串
mappings_str = "{\n"
for cmd, op in mappings.items():
for cmd, op in device_mappings.items():
mappings_str += f' "{cmd}": "{op}",\n'
# 添加默认映射（open1-open8, close1-close8）
for i in range(1, 9):
mappings_str += f' "open{i}": "open{i}",\n'
mappings_str += f' "close{i}": "close{i}",\n'
mappings_str += " }"
# 确定使用哪个 sender 脚本（每个设备独立的 sender）
sender_file = f"{global_config['base_dir']}/bin/sender_{device['id']}.py"
# 替换模板变量
content = template.format(
device_id=device['id'],
@@ -93,7 +113,9 @@ def generate_listener(device, global_config, mappings):
upc_port=device['upc_port'],
base_dir=global_config['base_dir'],
python_path=global_config['python_path'],
command_mappings=mappings_str
command_mappings=mappings_str,
sender_file=sender_file,
device_protocol=device_protocol.upper()
)
output_file = os.path.join(global_config['base_dir'], 'bin', f"{device['id']}.py")
@@ -101,14 +123,15 @@ def generate_listener(device, global_config, mappings):
f.write(content)
os.chmod(output_file, 0o755)
print(f"[生成] {output_file} ({protocol.upper()})")
print(f"[生成] {output_file} (监听:{listen_protocol.upper()}, 设备:{device_protocol.upper()})")
return output_file
def generate_crond(device, global_config):
"""生成保活检查文件"""
protocol = device.get('protocol', 'tcp').lower()
template_file = get_template_path(protocol, 'crond')
listen_protocol = device.get('listen_protocol', 'tcp').lower()
device_protocol = device.get('device_protocol', 'tcp').lower()
template_file = get_template_path(listen_protocol, 'crond')
# 检查模板文件是否存在
if not os.path.exists(template_file):
@@ -124,7 +147,8 @@ def generate_crond(device, global_config):
tms_server_ip=global_config['tms_server_ip'],
listen_port=device['listen_port'],
base_dir=global_config['base_dir'],
python_path=global_config['python_path']
python_path=global_config['python_path'],
device_protocol=device_protocol.upper()
)
output_file = os.path.join(global_config['base_dir'], 'crond', f"upcrond-{device['id']}.py")
@@ -132,14 +156,15 @@ def generate_crond(device, global_config):
f.write(content)
os.chmod(output_file, 0o755)
print(f"[生成] {output_file} ({protocol.upper()})")
print(f"[生成] {output_file} (监听:{listen_protocol.upper()}, 设备:{device_protocol.upper()})")
return output_file
def generate_sender(commands, global_config, protocol='tcp'):
"""生成发送模块文件"""
protocol = protocol.lower()
template_file = get_template_path(protocol, 'sender')
def generate_sender_for_device(device, command_set, global_config):
"""为指定设备生成独立的发送模块文件"""
device_id = device['id']
device_protocol = device.get('device_protocol', device.get('protocol', 'tcp')).lower()
template_file = get_template_path(device_protocol, 'sender')
# 检查模板文件是否存在
if not os.path.exists(template_file):
@@ -150,6 +175,7 @@ def generate_sender(commands, global_config, protocol='tcp'):
template = f.read()
# 构建指令定义字典字符串
commands = command_set.get('commands', {})
cmd_defs = "{\n"
for cmd, hex_str in commands.items():
cmd_defs += f' "{cmd}": "{hex_str}",\n'
@@ -159,31 +185,39 @@ def generate_sender(commands, global_config, protocol='tcp'):
command_definitions=cmd_defs
)
# 根据协议生成不同的文件名
output_file = os.path.join(global_config['base_dir'], 'bin', f'sender_{protocol}.py')
# 为每个设备生成独立的 sender 文件
output_file = os.path.join(global_config['base_dir'], 'bin', f'sender_{device_id}.py')
with open(output_file, 'w', encoding='utf-8') as f:
f.write(content)
os.chmod(output_file, 0o755)
print(f"[生成] {output_file} ({protocol.upper()})")
print(f"[生成] {output_file} ({device_protocol.upper()}, 指令集: {command_set.get('name', '未命名')})")
return output_file
def generate_all_senders(config, global_config):
"""为所有使用到的协议生成 sender 文件"""
commands = config['commands']
"""为所有设备生成独立的 sender 文件"""
command_sets = config.get('command_sets', {})
devices = config['devices']
# 收集所有使用的协议类型
protocols = set()
for device in devices:
if device.get('enabled', True):
protocol = device.get('protocol', 'tcp').lower()
protocols.add(protocol)
generated = []
for protocol in protocols:
generated.append(generate_sender(commands, global_config, protocol))
for device in devices:
if not device.get('enabled', True):
continue
# 获取设备使用的指令集
command_set_id = device.get('command_set', 'default')
if command_set_id not in command_sets:
# 兼容旧配置：如果找不到指令集，使用全局 commands
if 'commands' in config:
command_set = {'name': '全局指令', 'commands': config['commands']}
else:
print(f"[!] 警告: 设备 {device['id']} 指定的指令集 '{command_set_id}' 不存在")
continue
else:
command_set = command_sets[command_set_id]
generated.append(generate_sender_for_device(device, command_set, global_config))
return generated
@@ -220,13 +254,16 @@ def generate_control_script(devices, global_config):
device_list = []
for device in devices:
if device.get('enabled', True):
listen_protocol = device.get('listen_protocol', device.get('protocol', 'tcp')).upper()
device_protocol = device.get('device_protocol', device.get('protocol', 'tcp')).upper()
protocol_str = f"{listen_protocol}->{device_protocol}"
device_list.append({
'id': device['id'],
'name': device['name'],
'protocol': device.get('protocol', 'tcp').upper()
'protocol': protocol_str
})
# 构建设备列表字符串
# 构建设备列表字符串: ID 名称 协议
device_defs = "\n".join([
f" \"{d['id']}\" \"{d['name']}\" \"{d['protocol']}\""
for d in device_list
@@ -253,7 +290,8 @@ def check_config(config):
global_cfg = config.get('global', {})
devices = config.get('devices', [])
commands = config.get('commands', {})
command_sets = config.get('command_sets', {})
commands = config.get('commands', {}) # 兼容旧配置
mappings = config.get('mappings', {})
# 检查全局配置
@@ -269,6 +307,15 @@ def check_config(config):
device_ids = []
listen_ports = []
# 收集所有指令（用于验证映射）
all_commands = set()
if command_sets:
for set_id, cmd_set in command_sets.items():
if 'commands' in cmd_set:
all_commands.update(cmd_set['commands'].keys())
if commands:
all_commands.update(commands.keys())
for device in devices:
device_id = device.get('id')
if not device_id:
@@ -284,10 +331,23 @@ def check_config(config):
if field not in device:
errors.append(f"设备 {device_id} 缺少字段: {field}")
# 检查协议类型
protocol = device.get('protocol', 'tcp').lower()
if protocol not in ['tcp', 'udp']:
errors.append(f"设备 {device_id} 的协议类型无效: {protocol} (必须是 tcp 或 udp)")
# 检查协议类型（支持新旧配置兼容）
listen_protocol = device.get('listen_protocol', device.get('protocol', 'tcp')).lower()
device_protocol = device.get('device_protocol', device.get('protocol', 'tcp')).lower()
if listen_protocol not in ['tcp', 'udp']:
errors.append(f"设备 {device_id} 的监听协议类型无效: {listen_protocol} (必须是 tcp 或 udp)")
if device_protocol not in ['tcp', 'udp']:
errors.append(f"设备 {device_id} 的设备协议类型无效: {device_protocol} (必须是 tcp 或 udp)")
# 检查指令集
if 'command_set' in device:
command_set_id = device['command_set']
if command_set_id not in command_sets:
errors.append(f"设备 {device_id} 指定的指令集 '{command_set_id}' 不存在")
elif not commands and not command_sets:
warnings.append(f"设备 {device_id} 未指定指令集，且没有全局指令")
if 'listen_port' in device:
port = device['listen_port']
@@ -295,27 +355,36 @@ def check_config(config):
errors.append(f"设备 {device_id} 的监听端口 {port} 与其他设备冲突")
listen_ports.append(port)
# 检查指令定义
if not commands:
warnings.append("没有定义任何指令")
# 检查指令集
if not command_sets and not commands:
warnings.append("没有定义任何指令集或指令")
# 检查映射
for cmd, op in mappings.items():
if op not in commands and not any(op == f"open{i}" or op == f"close{i}" for i in range(1, 9)):
warnings.append(f"映射 '{cmd}' -> '{op}' 的目标指令未在 commands 中定义")
if op not in all_commands and not any(op == f"open{i}" or op == f"close{i}" for i in range(1, 9)):
warnings.append(f"映射 '{cmd}' -> '{op}' 的目标指令未在任何指令集中定义")
# 检查模板文件是否存在
protocols = set()
listen_protocols = set()
device_protocols = set()
for device in devices:
if device.get('enabled', True):
protocols.add(device.get('protocol', 'tcp').lower())
listen_protocols.add(device.get('listen_protocol', device.get('protocol', 'tcp')).lower())
device_protocols.add(device.get('device_protocol', device.get('protocol', 'tcp')).lower())
for protocol in protocols:
for template_type in ['listener', 'crond', 'sender']:
# 检查监听协议模板
for protocol in listen_protocols:
for template_type in ['listener', 'crond']:
template_file = get_template_path(protocol, template_type)
if not os.path.exists(template_file):
warnings.append(f"模板文件不存在: {template_file}")
# 检查设备协议模板（sender）
for protocol in device_protocols:
template_file = get_template_path(protocol, 'sender')
if not os.path.exists(template_file):
warnings.append(f"模板文件不存在: {template_file}")
# 输出检查结果
if errors:
print("\n[!] 配置错误:")
@@ -381,9 +450,11 @@ def main():
# 2. 为每个设备生成监听服务和保活脚本
print("\n[生成设备服务]")
command_sets = config.get('command_sets', {})
global_commands = config.get('commands', {})
for device in devices:
print(f"\n处理设备: {device['name']} ({device['id']}) - {device.get('protocol', 'tcp').upper()}")
generated_files.append(generate_listener(device, global_config, config['mappings']))
generated_files.append(generate_listener(device, global_config, config['mappings'], command_sets, global_commands))
generated_files.append(generate_crond(device, global_config))
# 3. 生成 crontab