300 lines
10 KiB
Python
300 lines
10 KiB
Python
import numpy as np
|
||
import copy
|
||
from typing import List
|
||
from FourChZreg_define import *
|
||
|
||
|
||
class AssemblyTemplateManager:
|
||
"""汇编指令模板管理器"""
|
||
|
||
def __init__(self, mk_instance, mk_instr, **kwargs):
|
||
self.mk = mk_instance
|
||
self.mk_instr = mk_instr
|
||
self.config_file = kwargs.get('config_file')
|
||
self.templates = {
|
||
'general_send_wait': self._generic_awg_control_template,
|
||
'ramp_fixed': self._ramp_mcu_fixed_template,
|
||
'ramp_step': self._ramp_mcu_template
|
||
}
|
||
|
||
def create_instructions(self, **kwargs):
|
||
template_type = kwargs.get('instr_type', str)
|
||
params = copy.deepcopy(kwargs)
|
||
return self.templates[template_type](**params)
|
||
|
||
def _codeword_encode(self, **kwargs):
|
||
sendc = kwargs.pop('sendc', 0)
|
||
wave_hold = kwargs.pop('wave_hold', 0)
|
||
ff_amp_index = kwargs.pop('ff_amp_index', 0)
|
||
fm_amp_index = kwargs.pop('fm_amp_index', 0)
|
||
bias_index = kwargs.pop('bias_index', 0)
|
||
fcw_index = kwargs.pop('fcw_index', 0)
|
||
pcw_index = kwargs.pop('pcw_index', 0)
|
||
code_clr = kwargs.pop('code_clr', 0)
|
||
env_index = kwargs.pop('env_index', 0)
|
||
|
||
codeword = 0
|
||
codeword |= sendc << 31
|
||
codeword |= wave_hold << 30
|
||
codeword |= ff_amp_index << 28
|
||
codeword |= fm_amp_index << 26
|
||
codeword |= bias_index << 24
|
||
codeword |= fcw_index << 22
|
||
codeword |= pcw_index << 19
|
||
codeword |= code_clr << 18
|
||
codeword |= env_index << 12
|
||
|
||
return codeword
|
||
|
||
def _codeword_gen(self, **kwargs):
|
||
codeword_configs = kwargs.pop('codeword_configs', [])
|
||
codeword_list = []
|
||
for config in codeword_configs:
|
||
codeword = self._codeword_encode(**config)
|
||
codeword_list.append(codeword)
|
||
return codeword_list
|
||
|
||
def write_register(self, address, value):
|
||
self.mk.rw_once('w', address, value, self.config_file)
|
||
|
||
def _generic_awg_control_template(self, **kwargs):
|
||
"""
|
||
通用 AWG汇编控制模版 (支持扫参)
|
||
"""
|
||
channel = kwargs.get('channel', 0)
|
||
codeword_list = self._codeword_gen(**kwargs)
|
||
send_interval_list = kwargs.get('send_interval', [100])
|
||
cycle_num = kwargs.get('cycle_num', 1)
|
||
sweep_config = kwargs.get('sweep_config', {})
|
||
sweep_num = sweep_config.get('sweep_num', 1)
|
||
sweep_offsets = sweep_config.get('offsets', [])
|
||
sweep_steps = sweep_config.get('steps', [])
|
||
sweep_reg_num = len(sweep_offsets)
|
||
|
||
dtcm_payload = [sweep_num, cycle_num, len(codeword_list), sweep_reg_num]
|
||
|
||
for cw, wait_clk in zip(codeword_list, send_interval_list):
|
||
dtcm_payload.append(cw)
|
||
dtcm_payload.append(wait_clk)
|
||
|
||
if sweep_reg_num > 0:
|
||
dtcm_payload.extend(sweep_offsets)
|
||
dtcm_payload.extend([s & 0xFFFFFFFF for s in sweep_steps])
|
||
|
||
dtcm_base = addr_base['DTCM0_BASE'] + channel * 0x600000
|
||
target_addr = dtcm_base + FourChZreg_define['mcu_reg']['DTFR'] + 4
|
||
self.write_register(target_addr, dtcm_payload)
|
||
|
||
return f"""
|
||
start:
|
||
lui x1 , 0x100
|
||
lui x2 , 0x200
|
||
|
||
addi x3 , x0 , 4
|
||
addi x4 , x0 , 22
|
||
addi x5 , x1 , 0x40
|
||
addi x6 , x2 , 0x40
|
||
|
||
load_nco_params_loop:
|
||
addi x4 , x4 , -1
|
||
lw x31, 0x00(x5)
|
||
sw x31, 0x00(x6)
|
||
add x5 , x5 , x3
|
||
add x6 , x6 , x3
|
||
bne x4 , x0 , load_nco_params_loop
|
||
|
||
lw x31, 0xb4(x1)
|
||
sw x31, 0xb4(x2)
|
||
|
||
lw x10, 0xdc(x1)
|
||
lw x11, 0xe0(x1)
|
||
lw x12, 0xe4(x1)
|
||
lw x17, 0xe8(x1)
|
||
|
||
addi x13, x1 , 0xec
|
||
slli x14, x12, 3
|
||
add x18, x13, x14
|
||
slli x14, x17, 2
|
||
add x20, x18, x14
|
||
|
||
run_sweep_iteration:
|
||
addi x28, x11, 0
|
||
middle_ensemble_loop:
|
||
addi x28, x28, -1
|
||
addi x26, x12, 0
|
||
addi x25, x13, 0
|
||
|
||
inner_wave_send_loop:
|
||
addi x26, x26, -1
|
||
lw x31, 0x00(x25)
|
||
lw x30, 0x04(x25)
|
||
addi x25, x25, 8
|
||
|
||
send x0 , x31, 0
|
||
|
||
bne x26, x0 , inner_wait_branch
|
||
beq x0 , x0 , outer_wait_branch
|
||
|
||
inner_wait_branch:
|
||
wait x0 , x30, -24
|
||
bne x26, x0 , inner_wave_send_loop
|
||
|
||
outer_wait_branch:
|
||
wait x0 , x30, -36
|
||
bne x28, x0 , middle_ensemble_loop
|
||
|
||
beq x10, x0 , mcu_exit
|
||
addi x10, x10, -1
|
||
wait x0 , x0 , 100
|
||
|
||
beq x17, x0 , run_sweep_iteration
|
||
addi x4 , x17, 0
|
||
addi x21, x18, 0
|
||
addi x23, x20, 0
|
||
|
||
update_param_loop:
|
||
addi x4 , x4 , -1
|
||
lw x29, 0(x21)
|
||
lw x24, 0(x23)
|
||
|
||
add x5 , x1 , x29
|
||
lw x31, 0(x5)
|
||
add x31, x31, x24
|
||
sw x31, 0(x5)
|
||
|
||
add x6 , x2 , x29
|
||
sw x31, 0(x6)
|
||
|
||
addi x21, x21, 4
|
||
addi x23, x23, 4
|
||
bne x4 , x0 , update_param_loop
|
||
|
||
beq x0 , x0 , run_sweep_iteration
|
||
|
||
mcu_exit:
|
||
exit x0 , x0 , 0
|
||
"""
|
||
|
||
def _ramp_mcu_template(self, **kwargs):
|
||
channel = kwargs.pop('channel')
|
||
ramp_mcu_registers = []
|
||
ramp_mcu_registers.append(0 << 16)
|
||
ramp_mcu_registers += [1 << 31]
|
||
param_num = kwargs.pop('param_num')
|
||
ensemble_num = kwargs.pop('ensemble_num')
|
||
ramp_mcu_registers.append(param_num)
|
||
ramp_mcu_registers.append(ensemble_num)
|
||
height_list = kwargs.pop('height', 0)
|
||
length_list = kwargs.pop('step_time', 0)
|
||
for height, length in zip(height_list, length_list):
|
||
ramp_mcu_registers += [height << 16]
|
||
ramp_mcu_registers += [length]
|
||
wait = 65536 / height * length
|
||
ramp_mcu_registers += [wait]
|
||
|
||
dtcm_base = addr_base['DTCM0_BASE'] + channel * 0x600000
|
||
target_addr = dtcm_base + FourChZreg_define['mcu_reg']['DTFR'] + 4
|
||
self.write_register(target_addr, ramp_mcu_registers)
|
||
|
||
return f"""
|
||
start:
|
||
lui x1 , 0x100
|
||
lui x2 , 0x200
|
||
lw x31, 0xdc(x1)
|
||
sw x31, 0xb8(x2)
|
||
lw x28, 0xe0(x1)
|
||
addi x6 , x0, 12
|
||
lw x7 , 0xe8(x1)
|
||
ensemble_loop:
|
||
addi x7 , x7, -1
|
||
addi x8 , x1, 0
|
||
lw x5 , 0xe4(x1)
|
||
ramp_loop:
|
||
addi x5, x5, -1
|
||
lw x31, 0xec(x8)
|
||
lw x30, 0xf0(x8)
|
||
lw x29, 0xf4(x8)
|
||
add x8 , x8 , x6
|
||
sw x30, 0xc0(x2)
|
||
sw x31, 0xbc(x2)
|
||
sw x28, 0xc4(x2)
|
||
wait x0 , x29, -30
|
||
bne x5 , x0 , ramp_loop
|
||
bne x7 , x0 , ensemble_loop
|
||
sw x0, 0xc4(x2)
|
||
exit x0, x0, 0
|
||
"""
|
||
|
||
def _ramp_mcu_fixed_template(self, **kwargs):
|
||
channel = kwargs.pop('channel')
|
||
ramp_mcu_registers = []
|
||
ramp_mcu_registers += [1 << 31]
|
||
ensemble_num = kwargs.pop('ensemble_num')
|
||
config_param_num = kwargs.pop('config_param_num')
|
||
ramp_mcu_registers.append(ensemble_num)
|
||
ramp_mcu_registers.append(config_param_num)
|
||
fixed_value_list = kwargs.pop('fixed_value')
|
||
wait_list = kwargs.pop('wait_clk')
|
||
for fixed_value, wait_clk in zip(fixed_value_list, wait_list):
|
||
ramp_mcu_registers += [fixed_value << 16 | 1 << 15]
|
||
ramp_mcu_registers += [wait_clk]
|
||
dtcm_base = addr_base['DTCM0_BASE'] + channel * 0x600000
|
||
target_addr = dtcm_base + FourChZreg_define['mcu_reg']['DTFR'] + 4
|
||
self.write_register(target_addr, ramp_mcu_registers)
|
||
|
||
return f"""
|
||
start:
|
||
lui x1 , 0x100
|
||
lui x2 , 0x200
|
||
lw x31, 0xdc(x1)
|
||
sw x31, 0xc4(x2)
|
||
addi x3 , x0, 8
|
||
lw x4 , 0xe0(x1)
|
||
ensemble_loop:
|
||
addi x4 , x4, -1
|
||
addi x6 , x1, 0
|
||
lw x5 , 0xe4(x1)
|
||
ramp_loop:
|
||
addi x5, x5, -1
|
||
lw x31, 0xe8(x6)
|
||
lw x30, 0xec(x6)
|
||
sw x31, 0xb8(x2)
|
||
add x6 , x6 , x3
|
||
bne x5 , x0 , ramp_loop_wait
|
||
jal x0 , ensemble_loop_wait
|
||
ramp_loop_wait:
|
||
wait x0 , x30, -24
|
||
jal x0 , ramp_loop
|
||
ensemble_loop_wait:
|
||
wait x0 , x30, -36
|
||
bne x4 , x0 , ensemble_loop
|
||
exit:
|
||
wait x0 , x0, 15
|
||
sw x0, 0xb8(x2)
|
||
sw x0, 0xc4(x2)
|
||
exit x0, x0, 0
|
||
"""
|
||
|
||
def _write_machine_codes_to_chip(self, machine_codes: str, **kwargs):
|
||
# 修正：同时把通道号传给 make_inst 的 channel_id 字段
|
||
channel = kwargs.pop('channel')
|
||
self.mk_instr.write(
|
||
machine_codes,
|
||
self.config_file,
|
||
channel_id=channel,
|
||
show=kwargs.pop('instr_show', False)
|
||
)
|
||
if 'inner_sync' in kwargs:
|
||
inner_sync = kwargs.pop('inner_sync')
|
||
if inner_sync:
|
||
self.write_register(addr_base['SYST_BASE'] + FourChZreg_define['sys_reg']['SYNCR'], 15 << 28 | 1 << 17)
|
||
self.mk.rw_once('r', addr_base['SYST_BASE'] + FourChZreg_define['pll_reg']['INTPLL_CLKRXPD'], [0] * 20,
|
||
self.config_file)
|
||
self.mk.rw_once('r', addr_base['DBGM_BASE'], [0] * 2048, self.config_file)
|
||
|
||
|
||
def instruction_config(mk_instance, mk_instr, **kwargs):
|
||
# 直接将 **kwargs 传入，__init__ 会自动匹配字典里的 'channel_id'，彻底避免重复传参报错
|
||
asm_templates = AssemblyTemplateManager(mk_instance, mk_instr, **kwargs)
|
||
machine_codes = asm_templates.create_instructions(**kwargs)
|
||
asm_templates._write_machine_codes_to_chip(machine_codes, **kwargs) |