class TestPRD extends Node: const DEVICE_CONSTANT: Dictionary = { 'PRD_N': { CONSTANTS = { 'POWER_UKP_1': 31, 'POWER_UKP_2': 63, 'DKM_BIT_2': 1, 'DKM_BIT_3': 2, 'BASE_ADDR_EMS_G': 0x120, 'ADDR_UG_LITERA_1': 0x106, 'ADDR_UG_LITERA_2': 0x108, 'ADDR_UG_LITERA_3': 0x10A, 'TEMP_THRESHOLD': 70, 'POWER_THRESHOLD': 40, 'RAY_DOU_NUM': 0x3, 'ATT_UM_2': 0x2, 'ATT_UM_3': 0x3, 'POWER_UM': 100, 'TIMER_DELAY': 10.0, 'STATE_MASHINE_TIMER': 0.3, }, PRIBOR_CONFIG = { '1н': {'angle': 0.0, 'sectors': [4, 8]}, '2н': {'angle': 90.0, 'sectors': [7, 11]}, '3н': {'angle': 180.0, 'sectors': [6, 10]}, '4н': {'angle': 270.0, 'sectors': [5, 9]}, }, FS_PARAMS = { 2: {'freq': 932.0, 'width': 100.0}, 3: {'freq': 1157.0, 'width': 100.0} }, START_LITERA = 2 }, 'PRD_V': { CONSTANTS = { 'POWER_UKP_1': 31, 'POWER_UKP_2': 63, 'DKM_BIT_4': 3, 'DKM_BIT_5': 4, 'BASE_ADDR_EMS_G': 0x120, 'ADDR_UG_LITERA_1': 0x106, 'ADDR_UG_LITERA_2': 0x108, 'ADDR_UG_LITERA_3': 0x10A, 'TEMP_THRESHOLD': 70, 'POWER_THRESHOLD': 40, 'RAY_DOU_NUM': 0x3, 'ATT_UM_4': 0x2, 'ATT_UM_5': 0x3, 'POWER_UM': 100, 'TIMER_DELAY': 10.0, 'STATE_MASHINE_TIMER': 0.3, }, PRIBOR_CONFIG = { '1в': {'angle': 0.0, 'sectors': [12, 16]}, '2в': {'angle': 90.0, 'sectors': [15, 19]}, '3в': {'angle': 180.0, 'sectors': [14, 18]}, '4в': {'angle': 270.0, 'sectors': [13, 17]}, }, FS_PARAMS = { 4: {'freq': 1227.0, 'width': 100.0}, 5: {'freq': 1575.0, 'width': 110.0}, }, START_LITERA = 4 }, 'PRD_K': { CONSTANTS = { 'POWER_UKP_1': 31, 'POWER_UKP_2': 63, 'DKM_BIT_1': 0, 'DKM_BIT_6': 5, 'DKM_BIT_7': 6, 'BASE_ADDR_EMS_G': 0x120, 'ADDR_UG_LITERA_1': 0x106, 'ADDR_UG_LITERA_2': 0x108, 'ADDR_UG_LITERA_3': 0x10A, 'TEMP_THRESHOLD': 40, 'POWER_THRESHOLD': 40, 'RAY_DOU_NUM': 0x3, 'ATT_UM_1': 0x0, 'ATT_UM_6': 0x3, 'ATT_UM_7': 0x2, 'POWER_UM': 100, 'TIMER_DELAY': 10.0, 'STATE_MASHINE_TIMER': 0.3, }, PRIBOR_CONFIG = { # Конфигурация параметров для каждого прибора '1к': {'angle': 0.0, 'sectors': [0, 20, 24]}, '2к': {'angle': 90.0, 'sectors': [3, 23, 27]}, '3к': {'angle': 180.0, 'sectors': [2, 22, 26]}, '4к': {'angle': 270.0, 'sectors': [1, 21, 25]}, }, FS_PARAMS = { # Конфигурация частот и ширин для каждого ФС 1: {'freq': 450.0, 'width': 80.0}, 6: {'freq': 2450.0, 'width': 110.0}, 7: {'freq': 4750.0, 'width': 110.0} }, START_LITERA = 1 } } var block_ip_config_n:= { 'maa2000_1': [0, false], 'maa2000_2': [1, false], 'ip9_50_1': [2, false], 'ip9_50_2': [3, false], 'ip12_50_1': [4, false], 'ip5_25': [7, false], } var block_kasseta_y5_prd_config_n: = { 'yau07': false, 'ems': false, 'ug': false, 'ukp1': false, 'ukp2': false, } # Результаты контроля var control_results_n: Dictionary = { 'block_ip': false, # Проверка касет П1 и ИП МА2000 'temperature': true, # Проверка температур УМ 'dkm_2': false, 'dkm_3': false, 'dou_2': false, 'dou_3': false, 'fs_2': 2, 'fs_3': 2, } #### var block_ip_config_v := { 'maa2000_1': [0, false], 'maa2000_2': [1, false], 'ip9_50_1': [2, false], 'ip9_50_2': [3, false], 'ip12_50_1': [4, false], 'ip5_25': [7, false], } var block_kasseta_y5_prd_config_v: = { 'yau07': false, 'ems': false, 'ug': false, 'ukp1': false, 'ukp2': false, } var control_results_v: Dictionary = { 'block_ip': false, # Проверка касет П1 и ИП МА2000 'temperature': true, # Проверка температур УМ 'dkm_4': false, 'dkm_5': false, 'dou_4': false, 'dou_5': false, 'fs_4': 2, 'fs_5': 2, } #### var block_kasseta_y5_prd_config_k: = { 'yau07': false, 'ems': false, 'ug': false, 'ukp1': false, 'ukp2': false, } var control_results_k: Dictionary = { 'block_ip': false, # Проверка касет П1 и ИП МА2000 'temperature': true, 'dkm_1': false, 'dkm_6': false, 'dkm_7': false, 'dou_6': false, 'dou_7': false, 'fs_1': 2, 'fs_6': 2, 'fs_7': 2, } var block_ip_config_k := { 'maa2000_1': [0, false], 'maa2000_2': [1, false], 'ip9_50_1': [2, false], 'ip9_50_2': [3, false], 'ip12_50_1': [4, false], 'ip9_50_3': [5, false], 'ip12_50_2': [6, false], 'ip5_25': [7, false], } #### const RAY_m5:= 'RAY_m5' const RAY_5:= 'RAY_5' const RAY_15:= 'RAY_15' const RAY_25:= 'RAY_25' const RAY_35:= 'RAY_35' const RAY_45:= 'RAY_45' enum DataIndices { STATUS_BOARD = 0, IN_BLANK_PRD = 1, IN_IMPULS_PRD = 7, OUT_BLANK_PRD = 13, OUT_IMPULS_PRD = 19, DRY_CONTACT = 27, DKM = 28, POWER_UKP_1 = 31, TEMPERATURE_UKP_1 = 47, POWER_UKP_2 = 63, TEMPERATURE_UKP_2 = 79, } 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 } const FS_KEYS := { 1: 'fs_1', 2: 'fs_2', 3: 'fs_3', 4: 'fs_4', 5: 'fs_5', 6: 'fs_6', 7: 'fs_7' } const TEMPERATURE_CONSTANTS := { 'CONST_MIN_TEMP': -25, 'MAXIMUM_CODE_ADC': 3796, 'MINIMUM_CODE_AD': 2660, 'TEMPERATURE_UKP_1': 47, 'TEMPERATURE_UKP_2': 79, } var power_for_litera: Dictionary = { 1: {RAY_m5: [], RAY_5: [], RAY_15: [], RAY_25: [], RAY_35: [], RAY_45: [],}, 2: {RAY_m5: [], RAY_5: [], RAY_15: [], RAY_25: [], RAY_35: [], RAY_45: [],}, 3: {RAY_m5: [], RAY_5: [], RAY_15: [], RAY_25: [], RAY_35: [], RAY_45: [],}, 4: {RAY_m5: [], RAY_5: [], RAY_15: [], RAY_25: [], RAY_35: [], RAY_45: [],}, 5: {RAY_m5: [], RAY_5: [], RAY_15: [], RAY_25: [], RAY_35: [], RAY_45: [],}, 6: {RAY_m5: [], RAY_5: [], RAY_15: [], RAY_25: [], RAY_35: [], RAY_45: [],}, 7: {RAY_m5: [], RAY_5: [], RAY_15: [], RAY_25: [], RAY_35: [], RAY_45: [],}, } var constants: Dictionary var pribor_config: Dictionary var fs_parameters: Dictionary var block_ip_config: Dictionary var block_kasseta_y5_config: Dictionary var control_results: 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 unit_prd: Object var power_amplify: Array var power_result: Dictionary #var control_label: Label var control_timer: Timer = Timer.new() var ems_6666_timer: Timer = Timer.new() var machine_timer: Timer = Timer.new() signal control_is_over() signal update_progress(progress_value: float) signal update_max_value_progress(max_progress_value: int) var cc: int = 0 var cmd_array: Array var progress_value: float: set(v): if progress_value != v: progress_value = v call_deferred('emit_signal', 'update_progress', v) var ecms: Dictionary ## Словарь помеховых воздействий ## Инициализация аргуметнов func _init(init_data: Dictionary, cmd_array_from_prd: Array) -> void: unit_prd = init_data['unit_prd'] cmd_array = cmd_array_from_prd func _ready() -> void: _determine_self_name() initialize_connections() setup_timer() func _process(_delta: float) -> void: pass ## Пакеты от ЯУ-07 func on_data_received(unit_pribor: Object) -> void: if machine_timer.is_stopped(): return var status: PackedByteArray = unit_pribor.status block_kasseta_y5_config = check_block_kasseta_y5(status, block_kasseta_y5_config) block_ip_config = check_block_ip(status, block_ip_config) control_results.temperature = check_temperature_um(status, power_amplify) socket_status_ems_g(status) _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 func socket_status_ems_g(packet_form_yau_07: PackedByteArray) -> void: 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(DataIndices.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(DataIndices.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(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) 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: for key in ecms.keys(): pass #print([fs_parameters, current_litera]) #print('ecms: %s %s' % [ecms[key].kni, ecms[key].params.freq]) if ecms.size() 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 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() ## В машине состояний вызов изменения состояния контроля progressbar_status() ## 'Выполняется инициализация ФС-%s' % current_litera func _handle_init_state() -> void: if current_litera in [1, 2, 3, 4, 5, 6, 7]: if not ecms.size(): state = STATE_MACHINE.FS_TURN_ON else: 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 _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 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 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')) signaller.connect('режим_работа', Callable(self, 'on_mode_changed').bind(false)) signaller.connect('режим_журнал', Callable(self, 'on_mode_changed').bind(false)) signaller.connect('режим_эмс', Callable(self, 'on_mode_changed').bind(false)) signaller.connect('режим_настройки', 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.INIT var current_unit = unit_prd.name var parts = current_unit.split('-') self_name = parts[-1] if self_name == '1н' or self_name == '2н' or self_name == '3н' or self_name == '4н': progress_value = 0 call_deferred('emit_signal', 'update_max_value_progress', 20) constants = DEVICE_CONSTANT.PRD_N.CONSTANTS pribor_config = DEVICE_CONSTANT.PRD_N.PRIBOR_CONFIG fs_parameters = DEVICE_CONSTANT.PRD_N.FS_PARAMS block_ip_config = block_ip_config_n block_kasseta_y5_config = block_kasseta_y5_prd_config_n control_results = control_results_n current_litera = DEVICE_CONSTANT.PRD_N.START_LITERA if self_name == '1в' or self_name == '2в' or self_name == '3в' or self_name == '4в': progress_value = 0 call_deferred('emit_signal', 'update_max_value_progress', 20) constants = DEVICE_CONSTANT.PRD_V.CONSTANTS pribor_config = DEVICE_CONSTANT.PRD_V.PRIBOR_CONFIG fs_parameters = DEVICE_CONSTANT.PRD_V.FS_PARAMS block_ip_config = block_ip_config_v block_kasseta_y5_config = block_kasseta_y5_prd_config_v control_results = control_results_v current_litera = DEVICE_CONSTANT.PRD_V.START_LITERA if self_name == '1к' or self_name == '2к' or self_name == '3к' or self_name == '4к': progress_value = 0 call_deferred('emit_signal', 'update_max_value_progress', 30) constants = DEVICE_CONSTANT.PRD_K.CONSTANTS pribor_config = DEVICE_CONSTANT.PRD_K.PRIBOR_CONFIG fs_parameters = DEVICE_CONSTANT.PRD_K.FS_PARAMS block_ip_config = block_ip_config_k block_kasseta_y5_config = block_kasseta_y5_prd_config_n control_results = control_results_k current_litera = DEVICE_CONSTANT.PRD_K.START_LITERA 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.TIMER_DELAY 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_mode_changed(in_tree: bool) -> void: if not in_tree: 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]]) else: pass func _process_dkm_status(status: PackedByteArray) -> void: if not control_results.has(_get_current_fs_key()): 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) func _get_current_fs_key() -> String: return FS_KEYS.get(current_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 _get_ray_key(machine_state: int) -> String: match machine_state: STATE_MACHINE.UG_SET_RAY_0: return RAY_m5 STATE_MACHINE.UG_SET_RAY_1: return RAY_5 STATE_MACHINE.UG_SET_RAY_2: return RAY_15 STATE_MACHINE.UG_SET_RAY_3: return RAY_25 STATE_MACHINE.UG_SET_RAY_4: return RAY_35 STATE_MACHINE.UG_SET_RAY_5: return RAY_45 _: return '' func on_stop_control() -> void: _handle_fs_turn_off_state() state = STATE_MACHINE.FS_TURN_OFF 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) 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, power_amplifier: Array) -> bool: const Constants: Dictionary = { 'CONST_MIN_TEMP': -25, 'MAXIMUM_CODE_ADC': 3796, 'MINIMUM_CODE_AD': 2660, 'TEMPERATURE_UKP_1': 47, 'TEMPERATURE_UKP_2': 79, } const TEMP: float = 115.0 / (Constants.MAXIMUM_CODE_ADC - Constants.MINIMUM_CODE_AD) var ukp1_online: int = (status_prd[0] >> 2) & 1 var ukp2_online = (status_prd[0] >> 3) & 1 var all_temp_is_ok: bool = true if ukp1_online: var tmp_1: int var temperature_ukp_1: int for i_temp in 6: 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) #power_amplifier[i_temp].value1 = tmp_1 if tmp_1>70: all_temp_is_ok = false if power_amplifier.size() > 12: # Темпуратура от УМ литеры 1 temperature_ukp_1 = status_prd.decode_u16(Constants.TEMPERATURE_UKP_1 + 2 * 7) tmp_1 = round(Constants.CONST_MIN_TEMP + (Constants.MAXIMUM_CODE_ADC - temperature_ukp_1) * TEMP + 3) #power_amplifier[6].value1 = tmp_1 if tmp_1>70: all_temp_is_ok = false if ukp2_online: for i_temp in 6: 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) if power_amplifier.size() > 12: #power_amplifier[i_temp+7].value1 = tmp_2 if tmp_2>70: all_temp_is_ok = false else: #power_amplifier[i_temp+6].value1 = tmp_2 if tmp_2>70: all_temp_is_ok = false return all_temp_is_ok ## Проверка ДКМ 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) -> Dictionary: var FS_RULES = [ {'key': 'fs_1', 'min': 0, 'max': 4}, {'key': 'fs_2', 'min': 4, 'max': 8}, {'key': 'fs_3', 'min': 8, 'max': 12}, {'key': 'fs_4', 'min': 12, 'max': 16}, {'key': 'fs_5', 'min': 16, 'max': 20}, {'key': 'fs_6', 'min': 20, 'max': 24}, {'key': 'fs_7', 'min': 24, 'max': 27} ] 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_key in FS_RULES.size(): var rule: Dictionary = FS_RULES[rule_key] 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 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 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_7', 'um_8', 'um_9', 'um_10', 'um_11', 'um_12'] ] else: um_names = [ ['um_1', 'um_2', 'um_3', 'um_4', 'um_5', 'um_6'], ['um_8', 'um_9', 'um_10', 'um_11', 'um_12', 'um_13'], ['um_7'] # Особый случай для третьего элемента ] 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