class_name TestPRD extends Node # Константы const CONFIG_PATH = "user://config/setting_prd_control.json" const DEFAULT_CONFIG_PATH = "res://config/setting_prd_control.json" const RAY_m5: String = "RAY_m5" const RAY_5: String = "RAY_5" const RAY_15: String = "RAY_15" const RAY_25: String = "RAY_25" const RAY_35: String = "RAY_35" const RAY_45: String = "RAY_45" const RAY_TYPES := [RAY_m5, RAY_5, RAY_15, RAY_25, RAY_35, RAY_45] enum STATE_MACHINE { INIT, FS_TURN_ON, FS_WORK, WRITE_ISA, READ_ISA, CHEK_KASSETA_Y5, UG_SET_RAY, UG_SET_RAY_0, UG_SET_RAY_1, UG_SET_RAY_2, UG_SET_RAY_3, UG_SET_RAY_4, UG_SET_RAY_5, IN_EMS_RUN, IN_EMS_STOP, FS_TURN_OFF, STOP_CONTROL, AWAIT } var ems_g_sock_dic: Dictionary = { 'ems_input_1': 0, 'ems_input_2': 0, 'ems_input_lit_1': 0, 'ems_output_1': 0, 'ems_output_2': 0, 'ems_output_lit_1': 0, } # Загруженные конфигурации var config_data: Dictionary var device_constants: Dictionary var base_configs: Dictionary var device_configs: Dictionary var data_indices: Dictionary var fs_keys: Dictionary var temperature_constants: Dictionary var ems_input_bits_config: Dictionary var fs_rules: Array var _ray_state_mapping: Dictionary # Переменные класса var constants: Dictionary var pribor_config: Dictionary var fs_parameters: Dictionary var block_ip_config: Dictionary var block_kasseta_y5_config: Dictionary var temperature_um: Dictionary var control_results: Dictionary var power_result: Dictionary var ecms: Dictionary var power_for_litera: Dictionary var self_name: String var state: int = STATE_MACHINE.INIT var current_litera: int var flag_write_isa: bool = false var flag_fs_status_ok: bool = false var flag_si_fs_input: bool = false var flag_si_fs_output: bool = false var flag_in_control: bool = false var flag_control_is_over: bool = false var unit_prd: Object var control_timer: Timer = Timer.new() var ems_6666_timer: Timer = Timer.new() var machine_timer: Timer = Timer.new() var cc: int = 0 var cmd_array: Array signal control_is_over() signal update_progress(progress_value: float) signal update_max_value_progress(max_progress_value: int) var progress_value: float: set(v): if progress_value != v: progress_value = v call_deferred('emit_signal', 'update_progress', v) ## Инициализация аргуметнов func _init(init_data: Dictionary, cmd_array_from_prd: Array) -> void: unit_prd = init_data['unit_prd'] cmd_array = cmd_array_from_prd _load_configuration() _initialize_ray_mapping() _initialize_power_structure() func _initialize_power_structure() -> void: power_for_litera = {} for litera in range(1, 8): power_for_litera[litera] = {} for ray_type in RAY_TYPES: power_for_litera[litera][ray_type] = [] func _ready() -> void: _determine_self_name() initialize_connections() setup_timer() func _get_ray_key(machine_state: int) -> String: return _ray_state_mapping.get(machine_state, "") func _ensure_config_exists() -> bool: DirAccess.make_dir_recursive_absolute("user://config") if FileAccess.file_exists(CONFIG_PATH): return true if not FileAccess.file_exists(DEFAULT_CONFIG_PATH): return false var src := FileAccess.open(DEFAULT_CONFIG_PATH, FileAccess.READ) var dst := FileAccess.open(CONFIG_PATH, FileAccess.WRITE) if src == null or dst == null: return false dst.store_string(src.get_as_text()) src.close() dst.close() return true func _load_configuration() -> void: if not _ensure_config_exists(): return var config_file = FileAccess.open(CONFIG_PATH, FileAccess.READ) if config_file: var json_text = config_file.get_as_text() config_file.close() var json = JSON.new() var parse_result = json.parse(json_text) if parse_result == OK: config_data = json.data _parse_config_data() else: push_error("Ошибка разбора JSON конфигурации: " + json.get_error_message()) else: push_error("Ошибка to load config file: " + CONFIG_PATH) func _parse_config_data() -> void: device_constants = _convert_numbers(config_data.get("device_constants", {})) base_configs = _convert_numbers(config_data.get("base_configs", {})) device_configs = _convert_numbers(config_data.get("device_configs", {})) data_indices = _convert_numbers(config_data.get("data_indices", {})) fs_keys = _convert_numbers(config_data.get("fs_keys", {})) temperature_constants = _convert_numbers(config_data.get("temperature_constants", {})) ems_input_bits_config = _convert_numbers(config_data.get("ems_input_bits_config", {})) fs_rules = _convert_numbers(config_data.get("fs_rules", [])) static func _convert_numbers(value): if value is float: # Если float без дробной части - конвертируем в int if value == int(value): return int(value) return value elif value is Dictionary: var result = {} for key in value: result[key] = _convert_numbers(value[key]) return result elif value is Array: var result = [] for item in value: result.append(_convert_numbers(item)) return result else: return value func _initialize_ray_mapping() -> void: _ray_state_mapping = { STATE_MACHINE.UG_SET_RAY_0: "RAY_m5", STATE_MACHINE.UG_SET_RAY_1: "RAY_5", STATE_MACHINE.UG_SET_RAY_2: "RAY_15", STATE_MACHINE.UG_SET_RAY_3: "RAY_25", STATE_MACHINE.UG_SET_RAY_4: "RAY_35", STATE_MACHINE.UG_SET_RAY_5: "RAY_45", } func _process(_delta: float) -> void: pass ## Пакеты от ЯУ-07 func on_data_received(unit_pribor: Object) -> void: if not flag_in_control: return var status: PackedByteArray = unit_pribor.status socket_status_ems_g(status) block_kasseta_y5_config = check_block_kasseta_y5(status, block_kasseta_y5_config) block_ip_config = check_block_ip(status, block_ip_config) temperature_um = check_temperature_um(status) if machine_timer.is_stopped(): return _process_dkm_status(status) if current_litera != 0: # Формирование мощностей в литерах var tek_litera: Dictionary = power_for_litera[current_litera] var interes = _get_ray_key(state) if interes != '': tek_litera[_get_ray_key(state)] = power_um_ukp(status, current_litera, tek_litera[_get_ray_key(state)]) power_for_litera[current_litera] = tek_litera # Методы для работы с JSON func _get_data_index(key: String) -> int: return data_indices.get(key, -1) func socket_status_ems_g(packet_form_yau_07: PackedByteArray) -> void: if self_name in ['1н', '2н', '3н', '4н']: ems_g_sock_dic.ems_input_1 = (packet_form_yau_07.decode_u16(_get_data_index('IN_BLANK_PRD')) & (1 << 9)) != 0 ems_g_sock_dic.ems_input_2 = (packet_form_yau_07.decode_u16(_get_data_index('IN_BLANK_PRD')) & (1 << 10)) != 0 if self_name in ['1в', '2в', '3в', '4в']: ems_g_sock_dic.ems_input_1 = (packet_form_yau_07.decode_u16(_get_data_index('IN_BLANK_PRD')) & (1 << 11)) != 0 ems_g_sock_dic.ems_input_2 = (packet_form_yau_07.decode_u16(_get_data_index('IN_BLANK_PRD')) & (1 << 12)) != 0 if self_name in ['1к', '2к', '3к', '4к']: ems_g_sock_dic.ems_input_lit_1 = (packet_form_yau_07.decode_u16(_get_data_index('IN_BLANK_PRD')) & (1 << 8)) != 0 ems_g_sock_dic.ems_input_1 = (packet_form_yau_07.decode_u16(_get_data_index('IN_BLANK_PRD')) & (1 << 13)) != 0 ems_g_sock_dic.ems_input_2 = (packet_form_yau_07.decode_u16(_get_data_index('IN_BLANK_PRD')) & (1 << 14)) != 0 ems_g_sock_dic.ems_output_1 = (packet_form_yau_07.decode_u16(_get_data_index('OUT_BLANK_PRD')) & (1 << 4)) != 0 ems_g_sock_dic.ems_output_2 = (packet_form_yau_07.decode_u16(_get_data_index('OUT_BLANK_PRD')) & (1 << 5)) != 0 ems_g_sock_dic.ems_output_lit_1 = (packet_form_yau_07.decode_u16(_get_data_index('OUT_BLANK_PRD')) & (1 << 3)) != 0 var number_socket_input: int if current_litera >= 1 and current_litera <= 7: number_socket_input = current_litera + 7 var in_blank: int = packet_form_yau_07.decode_u16(_get_data_index('IN_BLANK_PRD')) flag_si_fs_input = (in_blank & (1 << number_socket_input)) != 0 ## Флаг СИ от ФС входа в ЭМС-Г для текущей литеры var number_socket_output: int number_socket_output = 4 if current_litera in [2, 4, 6] else 3 if current_litera == 1 else 5 var out_blank: int = packet_form_yau_07.decode_u16(_get_data_index('OUT_BLANK_PRD')) flag_si_fs_output = (out_blank & (1 << number_socket_output)) != 0 ## Флаг Модуляции УМ ЭМС-Г для текущей литеры ## Пакеты от MP550Service func on_fs_update_state(packet_type_7: Dictionary, fs_mapping: Dictionary) -> void: var fs_key: String = fs_keys.get(str(current_litera), '') if fs_key.is_empty(): return var fs_result := {fs_key: 0} fs_result = on_fs_state(packet_type_7, fs_mapping, fs_result, fs_rules) control_results[fs_key] = fs_result[fs_key] flag_fs_status_ok = control_results[fs_key] == 1 ## func state_control_machine() -> void: match state: STATE_MACHINE.INIT: _handle_init_state() STATE_MACHINE.FS_TURN_ON: _handle_fs_turn_on_state() state = STATE_MACHINE.FS_WORK STATE_MACHINE.FS_WORK: var flag_ecm: bool = false for key in ecms: var params = ecms[key].params.freq if fs_parameters[current_litera].freq == params: var kni = ecms[key].kni if pribor_config[self_name].angle == kni: flag_ecm = true if flag_ecm and flag_fs_status_ok: ems_6666_timer.start() _handle_fs_work_state() STATE_MACHINE.CHEK_KASSETA_Y5: if block_kasseta_y5_config['yau07']: state = STATE_MACHINE.WRITE_ISA STATE_MACHINE.WRITE_ISA: if not flag_write_isa: _handle_ug_set_state(0x6) state = STATE_MACHINE.READ_ISA STATE_MACHINE.READ_ISA: if (unit_prd.cmd_state == unit_prd.CmdState.DONE) or (unit_prd.cmd_state == unit_prd.CmdState.FAIL): _handle_read_isa(unit_prd, [constants.ADDR_UG_LITERA_1, constants.ADDR_UG_LITERA_2, constants.ADDR_UG_LITERA_3], cmd_array) flag_write_isa = false state = STATE_MACHINE.UG_SET_RAY STATE_MACHINE.UG_SET_RAY: var addr: int = constants.ADDR_UG_LITERA_1 if current_litera == 3 or current_litera == 5 or current_litera == 7: addr = constants.ADDR_UG_LITERA_2 if unit_prd.isa_ports.has(addr) and (unit_prd.cmd_state == unit_prd.CmdState.DONE): if unit_prd.isa_ports[addr] & 0x07 == 0: state = STATE_MACHINE.UG_SET_RAY_1 elif unit_prd.isa_ports[addr] & 0x07 == 1: state = STATE_MACHINE.UG_SET_RAY_2 elif unit_prd.isa_ports[addr] & 0x07 == 2: state = STATE_MACHINE.UG_SET_RAY_3 elif unit_prd.isa_ports[addr] & 0x07 == 3: state = STATE_MACHINE.UG_SET_RAY_4 elif unit_prd.isa_ports[addr] & 0x07 == 4: state = STATE_MACHINE.UG_SET_RAY_5 elif unit_prd.isa_ports[addr] & 0x07 == 5: if not current_litera == 1: control_results = check_dou_result(control_results, power_for_litera, self_name, current_litera, constants) state = STATE_MACHINE.IN_EMS_STOP elif unit_prd.isa_ports[addr] & 0x07 == 6: state = STATE_MACHINE.UG_SET_RAY_0 if current_litera == 1: state = STATE_MACHINE.UG_SET_RAY_5 STATE_MACHINE.UG_SET_RAY_0: if flag_si_fs_input and not flag_write_isa: control_timer.stop() _handle_ug_set_state(0x0) await get_tree().create_timer(0.5).timeout state = STATE_MACHINE.READ_ISA STATE_MACHINE.UG_SET_RAY_1: if not flag_write_isa: _handle_ug_set_state(0x1) state = STATE_MACHINE.READ_ISA STATE_MACHINE.UG_SET_RAY_2: if not flag_write_isa: _handle_ug_set_state(0x2) state = STATE_MACHINE.READ_ISA STATE_MACHINE.UG_SET_RAY_3: if not flag_write_isa: _handle_ug_set_state(0x3) state = STATE_MACHINE.READ_ISA STATE_MACHINE.UG_SET_RAY_4: if not flag_write_isa: _handle_ug_set_state(0x4) state = STATE_MACHINE.READ_ISA STATE_MACHINE.UG_SET_RAY_5: if not flag_write_isa: _handle_ug_set_state(0x5) state = STATE_MACHINE.READ_ISA STATE_MACHINE.IN_EMS_STOP: ems_6666_timer.stop() state = STATE_MACHINE.FS_TURN_OFF STATE_MACHINE.FS_TURN_OFF: ems_6666_timer.stop() _handle_fs_turn_off_state() state = STATE_MACHINE.STOP_CONTROL STATE_MACHINE.STOP_CONTROL: _handle_stop_control_state() ## В машине состояний вызов изменения состояния прогресс бара контроля if block_kasseta_y5_config['yau07']: progressbar_status() ## 'Выполняется инициализация ФС-%s' % current_litera func _handle_init_state() -> void: if current_litera in [1, 2, 3, 4, 5, 6, 7]: var flag_ecm: bool = false for key in ecms: var params = ecms[key].params.freq if fs_parameters[current_litera].freq == params: var kni = ecms[key].kni if pribor_config[self_name].angle == kni: flag_ecm = true if not flag_ecm: state = STATE_MACHINE.FS_TURN_ON else: flag_control_is_over = true control_results = on_control_result(block_ip_config, control_results) power_result = prep_power(power_for_litera, self_name) emit_signal('control_is_over') machine_timer.stop() ##'Выполняется включение ФС-%s' % current_litera func _handle_fs_turn_on_state() -> void: manage_fs(true, self_name, pribor_config, fs_parameters, current_litera, constants.POWER_UM) control_timer.wait_time = constants.TIMER_DELAY control_timer.start() ##'Выполняется проверка ФС-%s' % current_litera func _handle_fs_work_state() -> void: if (current_litera == 2 and control_results.fs_2) or \ (current_litera == 3 and control_results.fs_3) or \ (current_litera == 4 and control_results.fs_4) or \ (current_litera == 5 and control_results.fs_5) or \ (current_litera == 6 and control_results.fs_6) or \ (current_litera == 7 and control_results.fs_7) or \ (current_litera == 1 and control_results.fs_1): state = STATE_MACHINE.CHEK_KASSETA_Y5 ## Подготовка к настройки и записи в УГ func _handle_ug_set_state(RAY_DOU_NUM: int) -> void: var dic_att := { 'RAY_DOU_NUM': RAY_DOU_NUM, 'ADDR_UG_LITERA_1': constants.ADDR_UG_LITERA_1, 'ADDR_UG_LITERA_2': constants.ADDR_UG_LITERA_2, 'ADDR_UG_LITERA_3': constants.ADDR_UG_LITERA_3, 'ATT_UM_3': 0x00, 'BASE_EMS_G': constants.BASE_ADDR_EMS_G } if self_name == '1к' or self_name == '2к' or self_name == '3к' or self_name == '4к': dic_att['ATT_UM_1'] = constants.ATT_UM_6 dic_att['ATT_UM_2'] = constants.ATT_UM_7 dic_att['ATT_UM_3'] = constants.ATT_UM_1 elif self_name == '1в' or self_name == '2в' or self_name == '3в' or self_name == '4в': dic_att['ATT_UM_1'] = constants.ATT_UM_4 dic_att['ATT_UM_2'] = constants.ATT_UM_5 elif self_name == '1н' or self_name == '2н' or self_name == '3н' or self_name == '4н': dic_att['ATT_UM_1'] = constants.ATT_UM_2 dic_att['ATT_UM_2'] = constants.ATT_UM_3 flag_write_isa = on_ug_setting(unit_prd, dic_att, cmd_array) ## Функция для продвижения статуса прогресс бара func progressbar_status() -> void: match state: STATE_MACHINE.INIT: cc += 10 if current_litera in [3, 5, 6, 7] else 0 if control_timer.get_time_left() > 0: progress_value = cc + control_timer.get_wait_time() - control_timer.get_time_left() if current_litera == 0: progress_value = 0 ##'Выполняется отключение ФС-%s' % current_litera func _handle_fs_turn_off_state() -> void: flag_fs_status_ok = false manage_fs(false, self_name, pribor_config, fs_parameters, current_litera, constants.POWER_UM) ##'Выполняется завершение проверки ФС-%s' % current_litera func _handle_stop_control_state() -> void: _advance_to_next_fs() state = STATE_MACHINE.INIT ## Переходит на следующую литеру, если 0 то считай закончил выполнения контроля func _advance_to_next_fs() -> void: match current_litera: 1: current_litera = 6 2: current_litera = 3 3: current_litera = 0 4: current_litera = 5 5: current_litera = 0 6: current_litera = 7 7: current_litera = 0 ###################################### func initialize_connections() -> void: var fs_mapping = _create_fs_mapping(self_name) signaller.conn(signaller.fs_update_state, Callable(self, 'on_fs_update_state').bind(fs_mapping)) signaller.conn(signaller.start_contol_device, Callable(self, '_on_start_contol_device')) interfer.connect('interfer_resized', Callable(self, 'on_interfer_resized')) unit_prd.connect('data_received', Callable(self, 'on_data_received')) var mode_signals = ['режим_работа', 'режим_журнал', 'режим_эмс', 'режим_настройки'] for signal_name in mode_signals: signaller.connect(signal_name, Callable(self, 'on_mode_changed').bind(false)) signaller.connect('режим_контроль', Callable(self, 'on_mode_changed').bind(true)) func on_interfer_resized(ecms_dic: Dictionary) -> void: ecms = ecms_dic func _on_start_contol_device()-> void: _determine_self_name() machine_timer.wait_time = constants.STATE_MASHINE_TIMER machine_timer.start() func _determine_self_name() -> void: cc = 0 state = STATE_MACHINE.get("INIT", 0) var current_unit = unit_prd.name var parts = current_unit.split('-') self_name = parts[-1] var device_type: String = _get_device_type() if device_type.is_empty(): return var device_config = device_constants.get(device_type, {}) var device_data_config = device_configs.get(device_type, {}) # Загружаем конфигурацию устройства constants = device_config.get("constants", {}) pribor_config = device_config.get("pribor_config", {}) # Конвертируем fs_params ключи в int var raw_fs_params = device_config.get("fs_params", {}) fs_parameters = {} for key in raw_fs_params: fs_parameters[int(key)] = raw_fs_params[key] # Загружаем конфигурации блоков block_ip_config = _load_block_config(device_data_config.get("block_ip", {}), "block_ip") block_kasseta_y5_config = _load_block_config(device_data_config.get("block_kasseta", {}), "block_kasseta") control_results = device_data_config.get("control_results", {}) current_litera = device_config.get("start_litera", 0) progress_value = 0 call_deferred('emit_signal', 'update_max_value_progress', device_config.get("progress_max", 20)) func _load_block_config(config, config_type: String) -> Dictionary: if config is String and config == "base": return base_configs.get(config_type, {}).duplicate(true) elif config is Dictionary: return config.duplicate(true) return {} func _get_device_type() -> String: if self_name.ends_with('н'): return 'PRD_N' elif self_name.ends_with('в'): return 'PRD_V' elif self_name.ends_with('к'): return 'PRD_K' return '' ## Настройка таймеров func setup_timer() -> void: add_child(control_timer) control_timer.timeout.connect(on_stop_control) control_timer.one_shot = true control_timer.wait_time = constants.get("TIMER_DELAY", 10.0) add_child(ems_6666_timer) ems_6666_timer.timeout.connect(_on_ems_6666_timer) ems_6666_timer.wait_time = 1.0 add_child(machine_timer) machine_timer.connect('timeout', Callable(self, 'on_state_mashine')) ## Таймер отправки команды в ЭМС-Г для модуляции func _on_ems_6666_timer() -> void: cmd_array.append([unit_prd.CmdCode.WRITE_ISA, [constants.BASE_ADDR_EMS_G, 6666]]) ## Таймер вызова машины состояний func on_state_mashine(): state_control_machine() ## При таймауте таймера ожидания выполнения команды и перехода на следующий шаг машины состояний func on_stop_control() -> void: _handle_fs_turn_off_state() state = STATE_MACHINE.FS_TURN_OFF ## Срабатывает при выходе или входе в режим контроль func on_mode_changed(in_tree: bool) -> void: flag_in_control = in_tree if not in_tree: ## TODO: Дополнительная проверка если контроль идет, если нет, то смысла вызывать нет flag_control_is_over = false progress_value = 0 machine_timer.stop() state = STATE_MACHINE.INIT _handle_fs_turn_off_state() #prd_v_module.cmd_array.append([unit_prd_v.CmdCode.WRITE_ISA, [constants_v.ADDR_UG_LITERA_1, ug_work_1, constants_v.ADDR_UG_LITERA_2, ug_work_2]]) func _process_dkm_status(status: PackedByteArray) -> void: if not control_results.has(_get_current_fs_key(current_litera, fs_keys)): return var dkm_bit = _get_dkm_bit_for_current_fs(current_litera, constants) if dkm_bit == 0: control_results['dkm_1'] = check_dkm(status, dkm_bit) var addr: int = constants.ADDR_UG_LITERA_1 if current_litera == 3 or current_litera == 5 or current_litera == 7: addr = constants.ADDR_UG_LITERA_2 if not unit_prd.isa_ports.has(addr) or not (unit_prd.cmd_state == unit_prd.CmdState.DONE): return if not unit_prd.isa_ports[addr] & 0x07 == 6: return if dkm_bit == 1: control_results['dkm_2'] = check_dkm(status, dkm_bit) elif dkm_bit == 2: control_results['dkm_3'] = check_dkm(status, dkm_bit) elif dkm_bit == 3: control_results['dkm_4'] = check_dkm(status, dkm_bit) elif dkm_bit == 4: control_results['dkm_5'] = check_dkm(status, dkm_bit) elif dkm_bit == 5: control_results['dkm_6'] = check_dkm(status, dkm_bit) elif dkm_bit == 6: control_results['dkm_7'] = check_dkm(status, dkm_bit) static func _get_current_fs_key(litera: int, fs_keys_dic: Dictionary) -> String: return fs_keys_dic.get(str(litera), '') ## Возвращает бит ДКМ для проверки текущей литеры static func _get_dkm_bit_for_current_fs(litera: int, constants_dic: Dictionary) -> int: match litera: 1: return constants_dic.DKM_BIT_1 2: return constants_dic.DKM_BIT_2 3: return constants_dic.DKM_BIT_3 4: return constants_dic.DKM_BIT_4 5: return constants_dic.DKM_BIT_5 6: return constants_dic.DKM_BIT_6 7: return constants_dic.DKM_BIT_7 _: return -1 ## Итоговое заключение по результам контроля всех ИП static func on_control_result(ip_config: Dictionary, ctrl_results: Dictionary) -> Dictionary: var all_blocks_ip_ok = true for block in ip_config: if not ip_config[block][1]: all_blocks_ip_ok = false ctrl_results['block_ip'] = all_blocks_ip_ok return ctrl_results ## Установка угла и аттенюации для УМ static func on_ug_setting(prd_unit: Object, settings_ug: Dictionary, prd_cmd_array: Array) -> bool: var result_1 = ((settings_ug.ATT_UM_1 & 0x07) << 3) | (settings_ug.RAY_DOU_NUM & 0x07) var result_2 = ((settings_ug.ATT_UM_2 & 0x07) << 3) | (settings_ug.RAY_DOU_NUM & 0x07) prd_cmd_array.append([prd_unit.CmdCode.WRITE_ISA, [ settings_ug.ADDR_UG_LITERA_1, result_1, settings_ug.ADDR_UG_LITERA_2, result_2, settings_ug.ADDR_UG_LITERA_3, settings_ug.ATT_UM_3, ] ]) return true static func _handle_read_isa(prd_unit: Object, read_isa_ports: Array, prd_cmd_array) -> void: prd_cmd_array.append([prd_unit.CmdCode.READ_ISA, read_isa_ports]) ## Управаление модулем фс (включить/отключить излучение) static func manage_fs(mode: bool, pribor: String, p_conf: Dictionary, fs_params: Dictionary, litera: int, power_um: int) -> void: # Получаем конфигурацию для текущего прибора var config: Dictionary = p_conf[pribor] var angle: float = config.angle var sectors: Array = config.sectors # Формируем выходные данные var interfer_name: String = 'k1' if mode else 'off' var fs_arr: Array = [] if litera in fs_params: var params = fs_params[litera] fs_arr.append([ angle, params.freq, params.width, sectors[litera - 1] if litera == 1 else sectors[litera - 5] if litera > 5 else sectors[litera - 4] if litera > 3 else sectors[litera - 2] ]) #signaller.emit_signal('emit_changed', mode) TODO: не отключаю подавление (даже если при входе в контроль он был отключен) signaller.emit_signal('interfer_init', interfer_name, fs_arr, power_um) ## Проверка касет П1 и ИП МА2000 static func check_block_ip(status_prd: PackedByteArray, ip_config: Dictionary) -> Dictionary: # Конфигурация проверяемых блоков var byte_dry: int = 27 var ug_valid: bool = (status_prd[0] >> 1) & 1 for block in ip_config: var bit: int = ip_config[block][0] if ug_valid: var status = status_prd[byte_dry] >> bit & 1 ip_config[block][1] = false if status else true return ip_config ## Проверка кассеты у-5 (ЯУ-07Б, ЭМС-Г, УГ, УКП-1, УКП-2) static func check_block_kasseta_y5(status_prd: PackedByteArray, y5_config: Dictionary) -> Dictionary: y5_config.yau07 = true y5_config.ems = (status_prd[0] & 1) and ((status_prd[0] >> 4) & 1) y5_config.ug = (status_prd[0] >> 1) & 1 y5_config.ukp1 = (status_prd[0] >> 2) & 1 y5_config.ukp2 = (status_prd[0] >> 3) & 1 return y5_config ## Мониторинг температуры static func check_temperature_um(status_prd: PackedByteArray) -> Dictionary: const Constants: Dictionary = { 'CONST_MIN_TEMP': -25, 'MAXIMUM_CODE_ADC': 3796, 'MINIMUM_CODE_AD': 2660, 'TEMPERATURE_UKP_1': 47, 'TEMPERATURE_UKP_2': 79, } const arr_um: Array = ['um_1', 'um_2', 'um_3', 'um_4', 'um_5', 'um_6', 'um_7', 'um_8', 'um_9', 'um_10', 'um_11', 'um_12', 'um_13', 'um_14', 'um_15', 'um_16'] const TEMP: float = 115.0 / (Constants.MAXIMUM_CODE_ADC - Constants.MINIMUM_CODE_AD) var return_dic_temp: Dictionary var ukp1_online: int = (status_prd[0] >> 2) & 1 var ukp2_online = (status_prd[0] >> 3) & 1 if ukp1_online: var tmp_1: int var temperature_ukp_1: int for i_temp in 8: temperature_ukp_1 = status_prd.decode_u16(Constants.TEMPERATURE_UKP_1 + 2 * i_temp) tmp_1 = round(Constants.CONST_MIN_TEMP + (Constants.MAXIMUM_CODE_ADC - temperature_ukp_1) * TEMP + 3) return_dic_temp[arr_um[i_temp]] = tmp_1 if ukp2_online: for i_temp in 8: var temperature_ukp_2: int = status_prd.decode_u16(Constants.TEMPERATURE_UKP_2 + 2 * i_temp) var tmp_2: int = round(Constants.CONST_MIN_TEMP + (Constants.MAXIMUM_CODE_ADC - temperature_ukp_2) * TEMP + 3) return_dic_temp[arr_um[i_temp+8]] = tmp_2 return return_dic_temp ## Проверка ДКМ static func check_dkm(status_prd: PackedByteArray, bit_dkm: int) -> bool: var dkm: int = 28 var status_dkm: bool = status_prd[dkm] >> bit_dkm & 1 return status_dkm ## Распределения номеров ФС по пирборам static func _create_fs_mapping(s_name: String) -> Dictionary: var suffixes: Dictionary = { 0: '1к', 1: '4к', 2: '3к', 3: '2к', 4: '1н', 5: '4н', 6: '3н', 7: '2н', 8: '1н', 9: '4н', 10: '3н', 11: '2н', 12: '1в', 13: '4в', 14: '3в', 15: '2в', 16: '1в', 17: '4в', 18: '3в', 19: '2в', 20: '1к', 21: '4к', 22: '3к', 23: '2к', 24: '1к', 25: '4к', 26: '3к', 27: '2к', } var mapping: Dictionary = {} for fs_num in suffixes: mapping[fs_num] = {'condition': s_name == suffixes[fs_num]} return mapping ## Получения статуса ФС static func on_fs_state(packet_type_7: Dictionary, fs_mapping: Dictionary, fs_results: Dictionary, fs_rules_config: Array) -> Dictionary: var updated_results := fs_results.duplicate() for fs_num in packet_type_7: if not (fs_num in fs_mapping and fs_mapping[fs_num]['condition']): continue var fs_data = packet_type_7[fs_num] for rule in fs_rules_config: if _should_apply_rule(fs_num, rule, updated_results): updated_results[rule.key] = fs_data.isg break return updated_results ## Вспомогательная функция для проверки применения правила static func _should_apply_rule(fs_num: int, rule: Dictionary, results: Dictionary) -> bool: return bool(fs_num >= rule.min and fs_num < rule.max) and results.has(rule.key) ## Получение мощности при выполнении контроля static func power_um_ukp(status: PackedByteArray, litera: int, pwr_result: Array) -> Array: const Constants: Dictionary = {'POWER_UKP_1' : 31, 'POWER_UKP_2': 63} var power_values: Array = [] var base_offset = 0 var iteration: int = 6 var ug_valid: bool = (status[0] >> 2) & 1 match litera: 1: # Обработка для current_fs = 1 (одно значение) var value = status.decode_u16(Constants.POWER_UKP_1 + 2 * 7) if ug_valid: power_values = [value] 2: # Обработка для current_fs = 2 (6 значений из УКП1) base_offset = Constants.POWER_UKP_1 if ug_valid: for i in iteration: power_values.append(status.decode_u16(base_offset + 2 * i)) 3: # Обработка для current_fs = 3 (6 значений из УКП2) base_offset = Constants.POWER_UKP_2 if ug_valid: for i in iteration: power_values.append(status.decode_u16(base_offset + 2 * i)) 4: # Обработка для current_fs = 4 (6 значений из УКП1) base_offset = Constants.POWER_UKP_1 if ug_valid: for i in iteration: power_values.append(status.decode_u16(base_offset + 2 * i)) 5: # Обработка для current_fs = 5 (6 значений из УКП2) base_offset = Constants.POWER_UKP_2 if ug_valid: for i in iteration: power_values.append(status.decode_u16(base_offset + 2 * i)) 6: # Обработка для current_fs = 6 (6 значений из УКП1) base_offset = Constants.POWER_UKP_1 if ug_valid: for i in iteration: power_values.append(status.decode_u16(base_offset + 2 * i)) 7: # Обработка для current_fs = 7 (6 значений из УКП2) base_offset = Constants.POWER_UKP_2 if ug_valid: for i in iteration: power_values.append(status.decode_u16(base_offset + 2 * i)) power_values = update_max_power_values(power_values, pwr_result) return power_values ## Обновление максимальных значений static func update_max_power_values(power_data: Array, max_power_values: Array) -> Array: if power_data.size() != 6: if max_power_values.is_empty(): max_power_values = power_data.duplicate() elif power_data[0] > max_power_values[0]: max_power_values = power_data.duplicate() return max_power_values if max_power_values.is_empty(): max_power_values = power_data.duplicate() else: for i in 6: if power_data[i] > max_power_values[i]: max_power_values[i] = power_data[i] return max_power_values.duplicate() ## Подготовка принятых данных по контролю для передачи в сцену static func prep_power(power_dic: Dictionary, prd_name: String)-> Dictionary: const RAY_KEYS := ["RAY_m5", "RAY_5", "RAY_15", "RAY_25", "RAY_35", "RAY_45"] var prep_pow_dic: Dictionary = {} var arr_litera: Array[int] match prd_name: '1н', '2н', '3н', '4н': arr_litera = [2, 3] '1в', '2в', '3в', '4в': arr_litera = [4, 5] '1к', '2к', '3к', '4к': arr_litera = [6, 7, 1] if not arr_litera: print_debug('Ошибка в prd_name') return prep_pow_dic var um_names: Array[Array] = [] if arr_litera.size() != 3: um_names = [ ['um_1', 'um_2', 'um_3', 'um_4', 'um_5', 'um_6'], ['um_9', 'um_10', 'um_11', 'um_12', 'um_13', 'um_14'] ] else: um_names = [ ['um_1', 'um_2', 'um_3', 'um_4', 'um_5', 'um_6'], ['um_9', 'um_10', 'um_11', 'um_12', 'um_13', 'um_14'], ['um_8'] ] for litera_index in arr_litera.size(): var litera: int = arr_litera[litera_index] var litera_data: Dictionary = power_dic.get(litera, {}) var um_data: Array[Array] = [[], [], [], [], [], []] for i in range(RAY_KEYS.size()): var ray_key: String = RAY_KEYS[i] var ray_array: Array = litera_data.get(ray_key, []) if ray_array.size() == 1: var single_value = ray_array[0] for j in range(6): um_data[j].append(single_value) else: for j in range(6): um_data[j].append(ray_array[j] if j < ray_array.size() else 0) var current_um_names: Array = um_names[litera_index] for um_index in range(current_um_names.size()): prep_pow_dic[current_um_names[um_index]] = um_data[um_index] return prep_pow_dic static func check_dou_result(control_dic: Dictionary, power_dic: Dictionary, pribor: String, litera: int, cnst: Dictionary) -> Dictionary: var all_dou_ok = true var dou: String = 'dou_2' if litera == 2 else 'dou_3' if pribor == '1к' or pribor == '2к' or pribor == '3к' or pribor == '4к': dou = 'dou_6' if litera == 6 else 'dou_7' elif pribor == '1в' or pribor == '2в' or pribor == '3в' or pribor == '4в': dou = 'dou_4' if litera == 4 else 'dou_5' var ray_dic: Dictionary = power_dic[litera] if ray_dic[RAY_m5].size() == 0 or \ ray_dic[RAY_5 ].size() == 0 or \ ray_dic[RAY_15].size() == 0 or \ ray_dic[RAY_25].size() == 0 or \ ray_dic[RAY_35].size() == 0 or \ ray_dic[RAY_45].size() == 0: control_dic[dou] = false return control_dic for i_pow_arr in ray_dic.values(): for i_pow in i_pow_arr.size(): if i_pow_arr[i_pow] < cnst.POWER_THRESHOLD: all_dou_ok = false control_dic[dou] = all_dou_ok return control_dic