maxim/uarep-ctl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

рефактор

Browse Source
This commit is contained in:
Maxim
2025-08-14 16:08:42 +03:00
parent 82a97d6b9e
commit aa8a714421
2 changed files with 314 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

415
scripts/pribor-prd.gd
View File

@@ -1,6 +1,8 @@
class TestPRD extends Node:
const RAY_25_1:= 'ray_25_1'
const RAY_25_6:= 'ray_25_6'
const RAY_25_7:= 'ray_25_7'
enum STATE_VAL { NONE = 0, GOOD = 1, ERROR = 2 }
# Конфигурация параметров для каждого прибора
enum STATE_MASHINE {
INIT,
FS_TURN_ON,
@@ -11,12 +13,28 @@ class TestPRD extends Node:
STOP_CONTROL,
AWAIT
}
var cur_ray: String = 'ray_25_1'
var power_result: Dictionary = {
'ray_25_1': [],
'ray_25_6': [],
'ray_25_7' : [],
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_result: Dictionary = {
RAY_25_1: [],
RAY_25_6: [],
RAY_25_7: [],
}
var constants: Dictionary
var pribor_config: Dictionary
var fs_parameters: Dictionary
@@ -24,28 +42,29 @@ class TestPRD extends Node:
var control_results: Dictionary
var self_name: String
var state: int = STATE_MASHINE.INIT
var control_label: Label
var current_fs: int = 1
var flag_check_dkm: bool
var flag_timer_started: bool
var ug_online: bool
var unit_prd: Object
var module_prd: Object
var power_amplify: Array
var fs_array_node: Array
var control_label: Label
var control_timer: Timer = Timer.new()
func _init(ctrlabel, u_prd, m_prd, constant, prd_config, fs_params, block_prd_config, power_amp, ctrl_results, fs_arr_node) -> void:
self.control_label = ctrlabel
self.unit_prd = u_prd
self.module_prd = m_prd
self.pribor_config = prd_config
self.fs_parameters = fs_params
self.constants = constant
self.block_config = block_prd_config
self.power_amplify = power_amp
self.control_results = ctrl_results
self.fs_array_node = fs_arr_node
func _init(init_data: Dictionary) -> void:
control_label = init_data['control_label']
power_amplify = init_data['power_amplifiers']
fs_array_node = init_data['fs_nodes']
unit_prd = init_data['unit_prd']
module_prd = init_data['module_prd']
constants = init_data['constants']
pribor_config = init_data['prd_config']
fs_parameters = init_data['fs_params']
block_config = init_data['block_ip_prd_config']
control_results = init_data['control_results']
func _ready() -> void:
initialize_connections()
@@ -59,117 +78,126 @@ class TestPRD extends Node:
return
var status: PackedByteArray = unit_pribor.status
ug_online = (status[0] >> 1) & 1
block_config = check_block_ip(status, block_config)
check_temperature_um(status, power_amplify)
if current_fs == 1 and control_results.fs_1:
control_results = check_dkm(status, constants.DKM_BIT_1, control_results)
flag_check_dkm = control_results.dkm_1
elif current_fs == 6 and control_results.fs_6:
control_results = check_dkm(status, constants.DKM_BIT_6, control_results)
flag_check_dkm = control_results.fs_6
elif current_fs == 7 and control_results.fs_7:
control_results = check_dkm(status, constants.DKM_BIT_7, control_results)
flag_check_dkm = control_results.dkm_7
if control_results.dkm_1 and current_fs == 1:
power_result['ray_25_1'] = power_um_ukp(status, current_fs, power_result['ray_25_1'])
if control_results.dkm_6 and current_fs == 6:
power_result['ray_25_6'] = power_um_ukp(status, current_fs, power_result['ray_25_6'])
if control_results.dkm_7 and current_fs == 7:
power_result['ray_25_7'] = power_um_ukp(status, current_fs, power_result['ray_25_7'])
_process_dkm_status(status)
func on_fs_update_state(packet_type_7: Dictionary, fs_mapping: Dictionary, fs_array: Array) -> void:
var fs_result: Dictionary
if current_fs == 1:
fs_result = {'fs_1': STATE_VAL.NONE}
elif current_fs == 6:
fs_result = {'fs_6': STATE_VAL.NONE}
elif current_fs == 7:
fs_result = {'fs_7': STATE_VAL.NONE}
var fs_key: String = FS_KEYS.get(current_fs, '')
if fs_key.is_empty():
return
var fs_result := {fs_key: STATE_VAL.NONE}
fs_result = on_fs_state(packet_type_7, fs_mapping, fs_result)
for fs in fs_array:
var fs_meta: String = fs.get_meta('fs')
if fs_result.has(fs_meta):
fs.state = fs_result[fs_meta]
control_results[fs_meta] = fs_result[fs_meta]
_update_fs_states(fs_result, fs_array)
func state_control_machine() -> void:
match state:
STATE_MASHINE.INIT:
control_label.text = 'Выполняется инициализация ФС-%s' % current_fs
if current_fs == 1 or current_fs == 6 or current_fs == 7:
state = STATE_MASHINE.FS_TURN_ON
else:
on_control_result()
state = STATE_MASHINE.AWAIT
_handle_init_state()
STATE_MASHINE.FS_TURN_ON:
control_label.text = 'Выполняется включение ФС-%s' % current_fs
manage_fs(true, self_name, pribor_config, fs_parameters, current_fs, constants.POWER_UM)
control_timer.wait_time = constants.TIMER_DELAY
control_timer.start()
state = STATE_MASHINE.FS_WORK
_handle_fs_turn_on_state()
STATE_MASHINE.FS_WORK:
control_label.text = 'Выполняется проверка ФС-%s' % current_fs
if (current_fs == 6 and flag_check_dkm and not flag_timer_started) or \
(current_fs == 7 and flag_check_dkm and not flag_timer_started) or \
(current_fs == 1 and not flag_timer_started):
state = STATE_MASHINE.UG_SET
_handle_fs_work_state()
STATE_MASHINE.UG_SET:
control_label.text = 'Выполняется настройка угла (%s) и атт (%s, %s, %s)' % [constants.dou_ray_nuber, constants.attenutor_um_1, constants.attenutor_um_2, constants.attenutor_um_3]
on_ug_setting()
state = STATE_MASHINE.UM_RUN
_handle_ug_set_state()
STATE_MASHINE.UM_RUN:
control_label.text = 'Выполняется настройка ЭМС'
module_prd.cmd_array.append([unit_prd.CmdCode.WRITE_ISA, [constants.BASE_ADDR_EMS_G, 6666]])
flag_timer_started = true
state = STATE_MASHINE.FS_WORK
flag_timer_started = _handle_um_run_state()
STATE_MASHINE.FS_TURN_OFF:
control_label.text = 'Выполняется отключение ФС-%s' % current_fs
manage_fs(false, self_name, pribor_config, fs_parameters, current_fs, constants.POWER_UM)
state = STATE_MASHINE.STOP_CONTROL
_handle_fs_turn_off_state()
STATE_MASHINE.STOP_CONTROL:
control_label.text = 'Выполняется завершение проверки ФС-%s' % current_fs
power_um_control_result()
flag_timer_started = false
flag_check_dkm = false
match current_fs:
1: current_fs = 6
6: current_fs = 7
7: current_fs = 0
state = STATE_MASHINE.INIT
_handle_stop_control_state()
func _handle_init_state() -> void:
control_label.text = 'Выполняется инициализация ФС-%s' % current_fs
if current_fs in [1, 2, 3, 4, 5, 6, 7]:
state = STATE_MASHINE.FS_TURN_ON
else:
on_control_result()
state = STATE_MASHINE.AWAIT
func _handle_fs_turn_on_state() -> void:
control_label.text = 'Выполняется включение ФС-%s' % current_fs
manage_fs(true, self_name, pribor_config, fs_parameters, current_fs, constants.POWER_UM)
control_timer.wait_time = constants.TIMER_DELAY
control_timer.start()
state = STATE_MASHINE.FS_WORK
func _handle_fs_work_state() -> void:
control_label.text = 'Выполняется проверка ФС-%s' % current_fs
if (current_fs == 2 and flag_check_dkm and not flag_timer_started) or \
(current_fs == 3 and flag_check_dkm and not flag_timer_started) or \
(current_fs == 4 and flag_check_dkm and not flag_timer_started) or \
(current_fs == 5 and flag_check_dkm and not flag_timer_started) or \
(current_fs == 6 and flag_check_dkm and not flag_timer_started) or \
(current_fs == 7 and flag_check_dkm and not flag_timer_started) or \
(current_fs == 1 and not flag_timer_started):
state = STATE_MASHINE.UG_SET
func _handle_ug_set_state() -> void:
control_label.text = 'Выполняется настройка угла (%s) и атт (%s, %s, %s)' % [
constants.RAY_DOU_NUM,
constants.ATT_UM_1,
constants.ATT_UM_6,
constants.ATT_UM_7,
]
on_ug_setting(module_prd, unit_prd, constants)
state = STATE_MASHINE.UM_RUN
func _handle_um_run_state() -> bool:
control_label.text = 'Выполняется настройка ЭМС'
module_prd.cmd_array.append([unit_prd.CmdCode.WRITE_ISA, [constants.BASE_ADDR_EMS_G, 6666]])
state = STATE_MASHINE.FS_WORK
return true
func _handle_fs_turn_off_state() -> void:
control_label.text = 'Выполняется отключение ФС-%s' % current_fs
manage_fs(false, self_name, pribor_config, fs_parameters, current_fs, constants.POWER_UM)
state = STATE_MASHINE.STOP_CONTROL
func _handle_stop_control_state() -> void:
control_label.text = 'Выполняется завершение проверки ФС-%s' % current_fs
control_results = power_um_control_result(control_results, constants, power_result)
flag_timer_started = false
flag_check_dkm = false
_advance_to_next_fs()
state = STATE_MASHINE.INIT
func _advance_to_next_fs() -> void:
match current_fs:
1: current_fs = 6
2: current_fs = 3
3: current_fs = 0
4: current_fs = 5
5: current_fs = 0
6: current_fs = 7
7: current_fs = 0
######################################
func initialize_connections() -> void:
_determine_self_name()
var fs_mapping = _create_fs_mapping(self_name)
signaller.conn(signaller.fs_update_state, Callable(self, 'on_fs_update_state').bind(fs_mapping, fs_array_node))
unit_prd.connect('data_received', Callable(self, 'on_data_received'))
func _determine_self_name() -> void:
var current_unit = unit_prd.name
var parts = current_unit.split('-')
self_name = parts[-1]
var fs_mapping = {
0: { 'condition': self_name == '1к' },
1: { 'condition': self_name == '4к' },
2: { 'condition': self_name == '3к' },
3: { 'condition': self_name == '2к' },
20: { 'condition':self_name == '1к' },
21: { 'condition': self_name == '4к' },
22: { 'condition': self_name == '3к' },
23: { 'condition': self_name == '2к' },
24: { 'condition': self_name == '1к' },
25: { 'condition': self_name == '4к' },
26: { 'condition': self_name == '3к' },
27: { 'condition': self_name == '2к' },
}
signaller.conn(signaller.fs_update_state, Callable(self, 'on_fs_update_state').bind(fs_mapping, fs_array_node))
unit_prd.connect('data_received', Callable(self, 'on_data_received'))
func setup_timer() -> void:
@@ -180,33 +208,92 @@ class TestPRD extends Node:
control_timer.start()
func on_control_result() -> void:
print('on_control_result')
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()
control_results = check_dkm(status, dkm_bit, control_results)
flag_check_dkm = control_results[_get_dkm_key_for_current_fs()]
if flag_check_dkm:
power_result[_get_ray_key_for_current_fs()] = power_um_ukp(status, current_fs, power_result[_get_ray_key_for_current_fs()])
func _get_current_fs_key() -> String:
return FS_KEYS.get(current_fs, '')
func _get_dkm_bit_for_current_fs() -> int:
match current_fs:
1: return constants.DKM_BIT_1
2: return constants.DKM_BIT_2
3: return constants.DKM_BIT_3
4: return constants.DKM_BIT_4
5: return constants.DKM_BIT_5
6: return constants.DKM_BIT_6
7: return constants.DKM_BIT_7
_: return -1
func _get_dkm_key_for_current_fs() -> String:
match current_fs:
1: return 'dkm_1'
2: return 'dkm_2'
3: return 'dkm_3'
4: return 'dkm_4'
5: return 'dkm_5'
6: return 'dkm_6'
7: return 'dkm_7'
_: return ''
func _get_ray_key_for_current_fs() -> String:
match current_fs:
1: return RAY_25_1
6: return RAY_25_6
7: return RAY_25_7
_: return ''
func _update_fs_states(fs_result: Dictionary, fs_array: Array) -> void:
for fs in fs_array:
var fs_meta: String = fs.get_meta('fs')
if fs_result.has(fs_meta):
var new_state = fs_result[fs_meta]
fs.state = new_state
control_results[fs_meta] = new_state
func on_stop_control() -> void:
state = STATE_MASHINE.FS_TURN_OFF
func on_ug_setting() -> void:
var result_1 = ((constants.attenutor_um_1 & 0x07) << 3) | (constants.dou_ray_nuber & 0x07)
var result_2 = ((constants.attenutor_um_2 & 0x07) << 3) | (constants.dou_ray_nuber & 0x07)
module_prd.cmd_array.append([unit_prd.CmdCode.WRITE_ISA, [constants.ADDR_UG_LITERA_1, result_1,
constants.ADDR_UG_LITERA_2, result_2,
constants.ADDR_UG_LITERA_3, constants.attenutor_um_3]])
module_prd.cmd_array.append([unit_prd.CmdCode.READ_ISA, [constants.ADDR_UG_LITERA_1, constants.ADDR_UG_LITERA_2, constants.ADDR_UG_LITERA_3]])
static func on_control_result() -> void:
print('on_control_result')
func power_um_control_result() -> void:
control_results['power_um'] = true
for amp in power_result[cur_ray]:
if amp > constants.TEMP_THRESHOLD:
control_results['power_um'] = false
break
static func on_ug_setting(prd_module: Object, prd_unit: Object, settings_ug: Dictionary) -> void:
var result_1 = ((settings_ug.ATT_UM_6 & 0x07) << 3) | (settings_ug.RAY_DOU_NUM & 0x07)
var result_2 = ((settings_ug.ATT_UM_7 & 0x07) << 3) | (settings_ug.RAY_DOU_NUM & 0x07)
prd_module.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_1]])
prd_module.cmd_array.append([prd_unit.CmdCode.READ_ISA, [settings_ug.ADDR_UG_LITERA_1, settings_ug.ADDR_UG_LITERA_2, settings_ug.ADDR_UG_LITERA_3]])
static func power_um_control_result(ctrl_results: Dictionary, constan: Dictionary, pwr_result: Dictionary) -> Dictionary:
ctrl_results['power_um'] = true
for amp_array in pwr_result.values():
for value in amp_array:
if value < constan.TEMP_THRESHOLD:
ctrl_results['power_um'] = false
break
return ctrl_results
## Управаление модулем фс (включить/отключить излучение)
static func manage_fs(mode: bool, pribor: String, p_conf: Dictionary, fs_params: Dictionary, cur_fs: int, power_um: int) -> void:
# Получаем конфигурацию для текущего прибора
var config: Dictionary = p_conf[pribor]
@@ -232,15 +319,15 @@ class TestPRD extends Node:
static func check_block_ip(status_prd: PackedByteArray, block_ip_config: Dictionary) -> Dictionary:
# Конфигурация проверяемых блоков
var byte_dry: int = 27
var ug_online: bool = (status_prd[0] >> 1) & 1
var ug_valid: bool = (status_prd[0] >> 1) & 1
for block in block_ip_config:
var bit: int = block_ip_config[block][0]
if ug_online:
if ug_valid:
var status = status_prd[byte_dry] >> bit & 1
block_ip_config[block][1] = false if status else true
return block_ip_config
## Проверка температур УМ от УКП
## Проверка температур УМ от УКП
static func check_temperature_um(status_prd: PackedByteArray, power_amplifier: Array) -> void:
const Constants: Dictionary = {
'CONST_MIN_TEMP': -25,
@@ -270,7 +357,6 @@ class TestPRD extends Node:
var tmp_2: int = round(Constants.CONST_MIN_TEMP + (Constants.MAXIMUM_CODE_ADC - temperature_ukp_2) * TEMP + 3)
power_amplifier[i_temp+7].value0 = tmp_2
## Проверка ДКМ
static func check_dkm(status_prd: PackedByteArray, bit_dkm: int, dkm_result: Dictionary) -> Dictionary:
var dkm: int = 28
@@ -284,51 +370,86 @@ class TestPRD extends Node:
return dkm_result
##
static func _create_fs_mapping(s_name: String) -> Dictionary:
var suffixes: Dictionary = {
0: 'к', 1: 'к', 2: 'к', 3: 'к',
4: 'н', 5: 'н', 6: 'н', 7: 'н',
8: 'н', 9: 'н', 10: 'н', 11: 'н',
12: 'в', 13: 'в', 14: 'в', 15: 'в',
16: 'в', 17: 'в', 18: 'в', 19: 'в',
20: 'к', 21: 'к', 22: 'к', 23: 'к',
24: 'к', 25: 'к', 26: 'к', 27: 'к',
}
var mapping: Dictionary = {}
for fs_num in suffixes:
var num_part = str((fs_num % 4) + 1)
mapping[fs_num] = {'condition': s_name == num_part + suffixes[fs_num]}
return mapping
##
static func on_fs_state(packet_type_7: Dictionary, fs_mapping: Dictionary, fs_results: Dictionary) -> Dictionary:
for fs in packet_type_7:
if fs in fs_mapping and fs_mapping[fs]['condition']:
var data = packet_type_7[fs]
if fs < 4 and fs_results.has('fs_1'):
fs_results['fs_1'] = data.isg
elif fs < 24 and fs_results.has('fs_6'):
fs_results['fs_6'] = data.isg
elif fs > 23 and fs_results.has('fs_7'):
fs_results['fs_7'] = data.isg
return fs_results
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]
# Проверяем все правила для текущего FS
for rule_key in FS_RULES:
var rule = 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.range[0] and fs_num < rule.range[1]) and results.has(rule.key)
##
static func power_um_ukp(status: PackedByteArray, crnt_fs: int, pwr_result: Array) -> Array:
const Constants: Dictionary = {'POWER_UKP_1' : 31, 'POWER_UKP_2': 63, 'TEMP_THRESHOLD': 40}
const Constants: Dictionary = {'POWER_UKP_1' : 31, 'POWER_UKP_2': 63}
var power_values: Array = []
var base_offset = 0
var is_data_valid = false
var iteration: int = 6
var ug_valid: bool = (status[0] >> 2) & 1
match crnt_fs:
1:
# Обработка для current_fs = 1 (одно значение)
var value = status.decode_u16(Constants.POWER_UKP_1 + 2 * 7)
power_values = [value]
is_data_valid = true
if ug_valid:
power_values = [value]
6:
# Обработка для current_fs = 6 (6 значений из УКП1)
base_offset = Constants.POWER_UKP_1
is_data_valid = (status[0] >> 2) & 1
if is_data_valid:
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
is_data_valid = (status[0] >> 2) & 1
if is_data_valid:
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