Files
imitator-case-y5/main.gd
2025-12-18 23:38:13 +03:00

525 lines
23 KiB
GDScript
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

extends Control
const CONFIG_PATH = "res://config/default_data.json"
const BUFFER_SIZE = 32
var udp_broadcast: UDPBroadcast
var uf_broadcast: UF_Broadcast
var default_data: Dictionary = {}
var dictionary_yau07: Dictionary
var dictionary_sockets_bit: Dictionary ## Байт 8 для отправки о состоянии ячеек
func _load_default_data():
var file = FileAccess.open(CONFIG_PATH, FileAccess.READ)
if file:
var json = JSON.new()
var error = json.parse(file.get_as_text())
if error == OK:
default_data = json.data["default_data"]
else:
push_error("JSON Parse Error: ", json.get_error_message())
file.close()
else:
push_error("Failed to load config file: ", CONFIG_PATH)
func _apply_default_data():
var prd_nodes = get_tree().get_nodes_in_group('prd')
for prd in prd_nodes:
var device_name = prd.get_meta('yau07')
if default_data.has(device_name):
var device_data = default_data[device_name]
_fill_ukp_fields(prd, device_data)
# Заполняем поля температур
func _fill_ukp_fields(prd_node: Control, device_data: Dictionary) -> void:
var device_name = prd_node.get_meta('yau07')
# Заполняем поля температур
var temp_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "temperature_ukp")
if device_data.has("temperature_ukp"):
for i in range(min(temp_fields.size(), device_data["temperature_ukp"].size())):
temp_fields[i].text = str(device_data["temperature_ukp"][i])
# Заполняем поля мощности
var power_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "power_ukp")
if device_data.has("power_ukp"):
for i in range(min(power_fields.size(), device_data["power_ukp"].size())):
power_fields[i].text = str(device_data["power_ukp"][i])
# Устанавливаем состояния сокетов
if device_data.has("sockets"):
var socket_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "socket")
for i in range(min(socket_fields.size(), device_data["sockets"].size())):
socket_fields[i].set_pressed_no_signal(device_data["sockets"][i])
# Устанавливаем состояния ДКМ
if device_data.has("dkm"):
var dkm_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "dkm")
for i in range(min(dkm_fields.size(), device_data["dkm"].size())):
dkm_fields[i].set_pressed_no_signal(device_data["dkm"][i])
# Устанавливаем состояния ДОУ2
if device_data.has("dou2"):
var dou2_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "dou2")
for i in range(min(dou2_fields.size(), device_data["dou2"].size())):
dou2_fields[i].set_pressed_no_signal(device_data["dou2"][i])
# Устанавливаем состояния ДОУ3
if device_data.has("dou3"):
var dou3_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "dou3")
for i in range(min(dou3_fields.size(), device_data["dou3"].size())):
dou3_fields[i].set_pressed_no_signal(device_data["dou3"][i])
# Устанавливаем состояния аттенюаторов 1
if device_data.has("att1"):
var att1_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "att1")
for i in range(min(att1_fields.size(), device_data["att1"].size())):
att1_fields[i].set_pressed_no_signal(device_data["att1"][i])
# Устанавливаем состояния аттенюаторов 2
if device_data.has("att2"):
var att2_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "att2")
for i in range(min(att2_fields.size(), device_data["att2"].size())):
att2_fields[i].set_pressed_no_signal(device_data["att2"][i])
# Устанавливаем состояния аттенюаторов 3
if device_data.has("att3"):
var att3_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "att3")
for i in range(min(att3_fields.size(), device_data["att3"].size())):
att3_fields[i].set_pressed_no_signal(device_data["att3"][i])
# Устанавливаем состояния источников питания (с инверсией)
if device_data.has("ip"):
var ip_fields = _get_ukp_nodes_for_device(device_name, [prd_node], "ip")
for i in range(min(ip_fields.size(), device_data["ip"].size())):
# Для IP используется инверсия: true в JSON = нажато = бит 0
var is_pressed = not device_data["ip"][i]
ip_fields[i].set_pressed_no_signal(is_pressed)
func _ready() -> void:
var device_yau07: Array = get_tree().get_nodes_in_group('device-addr')
for device in device_yau07:
device.connect('toggled', on_modify_device.bind(device.get_meta('device')))
var sockets: Array = get_tree().get_nodes_in_group('socket')
for soc in sockets:
soc.connect('toggled', _on_modify_sockets.bind(soc.get_meta('bit'), soc.get_meta('device'))) ## Мета bit от 0 до 3, мета ПРД
var power_ukp_arr_node: Array = get_tree().get_nodes_in_group('power_ukp')
for socket_power in power_ukp_arr_node:
socket_power.connect('text_submitted', _on_ukp_submitted.bind(socket_power.get_meta('device')))
var temperature_ukp_arr_node: Array = get_tree().get_nodes_in_group('temperature_ukp')
for socket_temp in temperature_ukp_arr_node:
socket_temp.connect('text_submitted', _on_ukp_submitted.bind(socket_temp.get_meta('device')))
var dkm_arr_node: Array = get_tree().get_nodes_in_group('dkm')
for dkm in dkm_arr_node:
dkm.connect('toggled', _on_modify_dkm.bind(dkm.get_meta('bit'), dkm.get_meta('device')))
var dou2_arr_node: Array = get_tree().get_nodes_in_group('dou2')
for dou2 in dou2_arr_node:
dou2.toggled.connect(_on_modify_status.bind(dou2.get_meta('bit'), dou2.get_meta('device'), "dou2"))
var dou3_arr_node: Array = get_tree().get_nodes_in_group('dou3')
for dou3 in dou3_arr_node:
dou3.toggled.connect(_on_modify_status.bind(dou3.get_meta('bit'), dou3.get_meta('device'), "dou3"))
var att1_arr_node: Array = get_tree().get_nodes_in_group('att1')
for att1 in att1_arr_node:
att1.toggled.connect(_on_modify_status.bind(att1.get_meta('bit'), att1.get_meta('device'), "att1"))
var att2_arr_node: Array = get_tree().get_nodes_in_group('att2')
for att2 in att2_arr_node:
att2.toggled.connect(_on_modify_status.bind(att2.get_meta('bit'), att2.get_meta('device'), "att2"))
var att3_arr_node: Array = get_tree().get_nodes_in_group('att3')
for att3 in att3_arr_node:
att3.toggled.connect(_on_modify_status.bind(att3.get_meta('bit'), att3.get_meta('device'), "att3"))
var ip_arr_node: Array = get_tree().get_nodes_in_group('ip')
for ip in ip_arr_node:
ip.connect('toggled', _on_modify_ip.bind(ip.get_meta('bit'), ip.get_meta('device')))
var ems_arr_node: Array = get_tree().get_nodes_in_group('ems_g')
for ems in ems_arr_node:
ems.connect('bits_changed', _on_modify_ems_g_status.bind(ems.get_meta('device')))
$uf_button.connect('toggled', on_modify_uf.bind($uf_button.get_meta('device')))
for bit in get_tree().get_nodes_in_group('uf_status'):
bit.connect('toggled', _on_update_uf_status.bind(bit.get_meta('bit')))
_update_tab_container()
_load_default_data()
_apply_default_data()
func _process(_delta: float) -> void:
for device in dictionary_yau07:
dictionary_yau07[device]._on_poll_timeout()
func _exit_tree():
for device in dictionary_yau07:
if dictionary_yau07[device]:
dictionary_yau07[device].close_udp_unit()
func on_modify_device(toggled: bool, name_device: String) -> void:
if toggled:
var logger_node: Node
for node in get_tree().get_nodes_in_group('logger'):
if node.get_meta('device') == name_device:
logger_node = node
break
udp_broadcast = UDPBroadcast.new(name_device, logger_node)
udp_broadcast.add_timers_to_scene(self)
udp_broadcast.start_broadcast()
dictionary_yau07[name_device] = udp_broadcast
print("Устройство %s активировано" % name_device)
dictionary_yau07[name_device].connect('update_isa_ports', _on_update_isa_ports)
dictionary_yau07[name_device].connect('update_socket_status', _on_update_socket_status)
var device_data = default_data[name_device]
_on_ukp_submitted('', name_device)
dictionary_yau07[name_device].status_manager.set_dou2_status(bool_array_to_number(device_data["dou2"]))
dictionary_yau07[name_device].status_manager.set_att2_status(bool_array_to_number(device_data["att2"]))
dictionary_yau07[name_device].status_manager.set_dou3_status(bool_array_to_number(device_data["dou3"]))
dictionary_yau07[name_device].status_manager.set_att3_status(bool_array_to_number(device_data["att3"]))
dictionary_yau07[name_device].set_ip_status(bool_array_to_number(device_data["ip"]))
dictionary_yau07[name_device].set_sockets_status(bool_array_to_number(device_data["sockets"]))
else:
if dictionary_yau07.has(name_device) and dictionary_yau07[name_device]:
dictionary_yau07[name_device].close_udp_unit()
dictionary_yau07.erase(name_device)
print("Устройство %s деактивировано" % name_device)
# Обновляем вкладки и переключаемся при активации
_update_tab_container(toggled, name_device)
func on_modify_uf(toggled: bool, name_device: String) -> void:
if toggled:
uf_broadcast = UF_Broadcast.new(name_device)
uf_broadcast.add_timers_to_scene(self)
uf_broadcast.start_broadcast()
dictionary_yau07[name_device] = uf_broadcast
print("Устройство %s активировано" % name_device)
else:
if dictionary_yau07.has(name_device) and dictionary_yau07[name_device]:
dictionary_yau07[name_device].close_udp_unit()
dictionary_yau07.erase(name_device)
print("Устройство %s деактивировано" % name_device)
func _update_tab_container(just_activated: bool = false, activated_device: String = ""):
var tab_container = $TabContainer
var device_nodes = get_tree().get_nodes_in_group('device-addr')
var has_visible_tabs = false
for i in range(tab_container.get_tab_count()):
var tab_control = tab_container.get_tab_control(i)
var device_name = tab_control.get_meta('yau07', '')
var device_button = null
for device in device_nodes:
if device.get_meta('device') == device_name:
device_button = device
break
if device_button:
var is_enabled = device_button.button_pressed
tab_container.set_tab_disabled(i, !is_enabled)
if is_enabled:
has_visible_tabs = true
if just_activated and device_name == activated_device:
tab_container.current_tab = i
tab_container.visible = has_visible_tabs
func _on_update_socket_status(status: int, device_name: String)->void:
var socket_arr = _get_nodes_for_device(device_name, 'socket')
for socket in socket_arr:
var bit_index = socket.get_meta('bit')
var is_bit_set = (status >> bit_index) & 1
socket.set_pressed_no_signal(is_bit_set == 1)
func _on_update_uf_status(toggled: bool, bit_index: int)->void:
if not dictionary_yau07.has('УФ'): return
var status = dictionary_yau07['УФ'].get_status_uf()
if bit_index == 6:
status = dictionary_yau07['УФ'].get_power_bit_uf()
if toggled: # Установить бит в 1
status = status | (1 << bit_index)
else: # Сбросить бит в 0
status = status & ~(1 << bit_index)
if bit_index == 6: dictionary_yau07['УФ'].set_power_bit_uf(status)
else: dictionary_yau07['УФ'].set_status_uf(status)
func _on_update_isa_ports(isa_ports: Dictionary, device_name: String)->void:
for port_addr in isa_ports:
var value = isa_ports[port_addr]
var port_configs = {
0x106: [ # Порт 0x106
{'group': 'dou2', 'mask': 0b111, 'shift': 0},
{'group': 'att2', 'mask': 0b111, 'shift': 3}
],
0x108: [ # Порт 0x108
{'group': 'dou3', 'mask': 0b111, 'shift': 0},
{'group': 'att3', 'mask': 0b111, 'shift': 3}
],
0x10A: [ # Порт 0x10A
{'group': 'att1', 'mask': 0b111, 'shift': 0}
]
}
if port_configs.has(port_addr):
for config in port_configs[port_addr]:
var status = (value >> config.shift) & config.mask
_update_specific_device_buttons(config.group, status, device_name)
func _update_specific_device_buttons(group_name: String, status: int, target_device: String) -> void:
"""Обновление кнопок только для конкретного устройства"""
var buttons = _get_nodes_for_device(target_device, group_name)
for button in buttons:
if button.has_meta('bit'):
var bit_value = button.get_meta('bit')
var should_be_pressed = (status == bit_value)
if button.button_pressed != should_be_pressed:
button.set_pressed_no_signal(should_be_pressed)
if should_be_pressed:
button.grab_focus()
func _get_nodes_for_device(device: String, group_name: String) -> Array:
"""Получение всех нод в группе, принадлежащих указанному устройству"""
var prd_nodes = get_tree().get_nodes_in_group('prd')
var result = []
for prd in prd_nodes:
if prd.get_meta('yau07') == device:
for node in get_tree().get_nodes_in_group(group_name):
if prd.is_ancestor_of(node) and node.get_meta('device') == device:
result.append(node)
break
return result
## Функция для установки статусов ячеек на связи (УГ, ЭМС-Г, УКП)
func _on_modify_sockets(toggled: bool, bit: int, device: String)->void:
if dictionary_yau07.has(device):
var udp_unit = dictionary_yau07[device]
var current_status = udp_unit.get_sockets_status() # Используем геттер
var new_status = _set_bit(current_status, bit, toggled)
udp_unit.set_sockets_status(new_status) # Используем сеттер
print("Статус сокетов устройства %s обновлен: 0x%02X" % [device, new_status])
## Функция для установки статусов ДКМ в ПРД
func _on_modify_dkm(toggled: bool, bit: int, device: String)->void:
if dictionary_yau07.has(device):
var udp_unit = dictionary_yau07[device]
var current_status = udp_unit.get_dkm_status() # Используем геттер
var new_status = _set_bit(current_status, bit, toggled)
udp_unit.set_dkm_status(new_status) # Используем сеттер
print("Статус DKM устройства %s обновлен: 0x%02X" % [device, new_status])
func _on_modify_ems_g_status(input_impulse: int, input_const: int, output_impulse: int, output_const: int, device: String)->void:
if not dictionary_yau07.has(device):
return
var udp_unit = dictionary_yau07[device]
udp_unit.set_ems_g_status(input_impulse, input_const, output_impulse, output_const)
func _on_modify_status(toggled: bool, bit: int, device: String, status_type: String) -> void:
if not dictionary_yau07.has(device):
return
var udp_unit = dictionary_yau07[device]
var button_group = _get_nodes_for_device(device, status_type)
if not toggled:
# Восстанавливаем предыдущее состояние
for button in button_group:
if button.has_meta('bit') and button.get_meta('bit') == bit:
button.set_pressed(true)
return
# Выключаем другие кнопки в группе
for button in button_group:
if button.has_meta('bit') and button.get_meta('bit') != bit and button.button_pressed:
button.set_pressed_no_signal(false)
# Устанавливаем статус через менеджер
match status_type:
"dou2": udp_unit.status_manager.set_dou2_status(bit)
"att1": udp_unit.status_manager.set_att1_status(bit)
"att2": udp_unit.status_manager.set_att2_status(bit)
"dou3": udp_unit.status_manager.set_dou3_status(bit)
"att3": udp_unit.status_manager.set_att3_status(bit)
func _on_modify_ip(toggled: bool, bit: int, device: String):
if dictionary_yau07.has(device):
var udp_unit = dictionary_yau07[device]
var current_status = udp_unit.get_ip_status()
# Инверсия: нажата = 0, отжата = 1
var bit_value = 0 if toggled else 1
var new_status
if bit_value == 0: new_status = current_status & ~(1 << bit) # Устанавливаем бит в 0
else: new_status = current_status | (1 << bit) # Устанавливаем бит в 1
udp_unit.set_ip_status(new_status)
print("Статус ИП устройства %s обновлен: 0x%02X" % [device, new_status])
## Функция для установки значений в ячейках УКП
func _on_ukp_submitted(_new_text: String, device: String) -> void:
if not dictionary_yau07.has(device):
return
var prep_temperature_ukp1_array: Array = _on_text_temperature_changed(device, 1)
var prep_temperature_ukp2_array: Array = _on_text_temperature_changed(device, 2)
var prep_power_ukp1_array: Array = _on_text_power_changed(device, 1)
var prep_power_ukp2_array: Array = _on_text_power_changed(device, 2)
var combine_data_ukp1: Array = _combine_arrays(prep_temperature_ukp1_array, prep_power_ukp1_array)
var combine_data_ukp2: Array = _combine_arrays(prep_temperature_ukp2_array, prep_power_ukp2_array)
_form_buffers(device, combine_data_ukp1, 1, dictionary_yau07)
_form_buffers(device, combine_data_ukp2, 2, dictionary_yau07)
func _on_text_temperature_changed(device: String, number_device: int) -> Array:
return _process_ukp_data(device, number_device, "temperature_ukp", _convert_temperature)
func _on_text_power_changed(device: String, number_device: int) -> Array:
return _process_ukp_data(device, number_device, "power_ukp", _convert_power)
func _process_ukp_data(device: String, number_device: int, group_name: String, converter: Callable) -> Array:
var prd_module_arr = get_tree().get_nodes_in_group('prd')
var current_change_nodes: Array = _get_ukp_nodes_for_device(device, prd_module_arr, group_name)
var result_arr: Array = [0xffff, 0xffff, 0xffff, 0xffff, 0xffff, 0xffff, 0xffff, 0xffff]
var config: Dictionary = _get_ukp_config(number_device)
for change_node in current_change_nodes:
var place = change_node.get_meta('place')
if change_node.get_meta('place') in config.meta_range:
if number_device == 2: place -= 8
var value = _safe_convert_to_int(change_node.text)
result_arr[place] = converter.call(value)
return result_arr
static func _get_ukp_nodes_for_device(device: String, prd_module_arr: Array, group_name: String) -> Array:
for prd in prd_module_arr:
if device == prd.get_meta('yau07'):
var nodes = []
for node in prd.get_tree().get_nodes_in_group(group_name):
if prd.is_ancestor_of(node):
nodes.append(node)
return nodes
return []
static func _get_ukp_config(number_device: int) -> Dictionary:
if number_device == 1:
return {
"meta_range": [0, 1, 2, 3 ,4, 5, 6, 7]
}
else:
return {
"meta_range": [8, 9, 10, 11, 12, 13, 14, 15]
}
static func _convert_temperature(value: int) -> int:
const CONST_MIN_TEMP: int = -25
const MAXIMUM_CODE_ADC: int = 3796
const MINIMUM_CODE_AD: int = 2660
const TEMP: float = 115.0 / (MAXIMUM_CODE_ADC - MINIMUM_CODE_AD)
var result: float = MAXIMUM_CODE_ADC - (value - CONST_MIN_TEMP - 3) / TEMP
return int(result)
static func _convert_power(value: int) -> int:
return value
static func _safe_convert_to_int(text: String) -> int:
text = text.strip_edges().to_lower()
if text.is_valid_int():
return int(text)
elif text.is_valid_float():
return roundi(float(text))
elif text.begins_with("0x") and text.substr(2).is_valid_hex_number():
return text.hex_to_int()
else:
return 0
static func _combine_arrays(power_arr: Array, temp_arr: Array)-> Array:
return temp_arr + power_arr
static func _form_buffers(device: String, ukp_data: Array, number_device: int, dic_yau07: Dictionary):
var buffer: PackedByteArray = PackedByteArray()
buffer.resize(BUFFER_SIZE)
for i in ukp_data.size():
buffer.encode_u16(2 * i, ukp_data[i])
if dic_yau07.has(device):
var udp_unit = dic_yau07[device]
if number_device == 1:
udp_unit.ukp0_data = buffer
print("Данные UKP0 устройства %s обновлены" % device)
elif number_device == 2:
udp_unit.ukp1_data = buffer
print("Данные UKP1 устройства %s обновлены" % device)
static func _set_bit(bits: int, index: int, val: bool) -> int:
return bits | (1 << index) if val else bits & ~(1 << index)
static func bool_array_to_number(bool_array: Array)-> int:
var result: int = 0
for i in range(bool_array.size()):
if bool_array[i]:
result |= (1 << i)
return result
## Метод для работы с ISA портами
func read_isa_ports(device: String, port_addresses: Array) -> Dictionary:
"""Чтение ISA портов указанного устройства"""
if dictionary_yau07.has(device):
return dictionary_yau07[device].read_isa_ports(port_addresses)
else:
push_error("Устройство %s не найдено" % device)
return {}
## Метод для записи ISA портов
func write_isa_port(device: String, port_address: int, value: int) -> bool:
"""Запись значения в ISA порт указанного устройства"""
if dictionary_yau07.has(device):
return dictionary_yau07[device].set_isa_port(port_address, value)
else:
push_error("Устройство %s не найдено" % device)
return false