Save current local changes

device_commands.py

@@ -18,6 +18,10 @@ DS1809_CMD_TOTAL_LENGTH = 10 # Total bytes when sending DS1809 UC/DC pulse comma
 DS1809_CMD_HEADER = "AAAA"
 STM32_DAC_CMD_TOTAL_LENGTH = 10 # Total bytes when sending STM32 DAC command
 STM32_DAC_CMD_HEADER = "BBBB"
+AD9102_WAVE_CTRL_TOTAL_LENGTH = 10 # Total bytes when sending AD9102 waveform control command
+AD9102_WAVE_CTRL_HEADER = "CCCC"
+AD9102_WAVE_DATA_TOTAL_LENGTH = 30 # Total bytes when sending AD9102 waveform data packet
+AD9102_WAVE_DATA_HEADER = "DDDD"
 AD9102_SAW_STEP_DEFAULT = 1
 AD9102_PAT_PERIOD_DEFAULT = 0xFFFF
 AD9102_PAT_PERIOD_BASE_DEFAULT = 0x02
@@ -27,6 +31,13 @@ AD9102_FLAG_SRAM_FMT = 0x0008
 AD9102_SRAM_SAMPLES_DEFAULT = 16
 AD9102_SRAM_HOLD_DEFAULT = 1
 AD9102_SRAM_AMP_DEFAULT = 8191
+AD9102_WAVE_OPCODE_BEGIN = 0x0001
+AD9102_WAVE_OPCODE_COMMIT = 0x0002
+AD9102_WAVE_OPCODE_CANCEL = 0x0003
+AD9102_WAVE_CHUNK_SAMPLES = 12
+AD9102_SAMPLE_MIN = -8192
+AD9102_SAMPLE_MAX = 8191
+AD9102_WAVE_MAX_SAMPLES = 4096
 AD9833_FLAG_ENABLE = 0x0001
 AD9833_FLAG_TRIANGLE = 0x0002
 AD9833_MCLK_HZ_DEFAULT = 20_000_000
@@ -219,17 +230,45 @@ def send_STM32_DAC(prt, bytestring):
     print("Sent: STM32 DAC command.")
 
 
+def send_AD9102_waveform_control(prt, bytestring, quiet: bool = False):
+    ''' Control custom AD9102 SRAM upload (0xCCCC + ...).
+        Expected device answer: STATE.
+    '''
+    if len(bytestring) != AD9102_WAVE_CTRL_TOTAL_LENGTH:
+        print("Error. Wrong parameter string for AD9102 waveform control command.")
+        return None
+    prt.write(bytestring)
+    if not quiet:
+        print("Sent: AD9102 waveform control command.")
+
+
+def send_AD9102_waveform_data(prt, bytestring, quiet: bool = False):
+    ''' Send custom AD9102 SRAM samples (0xDDDD + ...).
+        Expected device answer: STATE.
+    '''
+    if len(bytestring) != AD9102_WAVE_DATA_TOTAL_LENGTH:
+        print("Error. Wrong parameter string for AD9102 waveform data command.")
+        return None
+    prt.write(bytestring)
+    if not quiet:
+        print("Sent: AD9102 waveform data command.")
+
+
 # ---- Getting data
 
 
-def get_STATE(prt):
+def get_STATE(prt, quiet: bool = False):
     ''' Get decoded state of the device in byte format (2 bytes).
     '''
 
-    print("Received "+str(prt.inWaiting())+" bytes.")
+    if not quiet:
+        print("Received "+str(prt.inWaiting())+" bytes.")
     if prt.inWaiting()!=2:
-        print("Error. Couldn't get STATE data. prt.inWaiting():", prt.inWaiting())
-        print("Flushing input data:", prt.read(prt.inWaiting()))        
+        if not quiet:
+            print("Error. Couldn't get STATE data. prt.inWaiting():", prt.inWaiting())
+            print("Flushing input data:", prt.read(prt.inWaiting()))
+        else:
+            prt.read(prt.inWaiting())
         # print("Flushing input data:", prt.read(2), prt.read(2))
         return None
 
@@ -534,6 +573,61 @@ def create_STM32_DAC_command(dac_code: int, enable: bool = True):
     return bytearray.fromhex(data)
 
 
+def create_AD9102_waveform_control_command(opcode: int, param0: int = 0, param1: int = 0):
+    if opcode is None:
+        opcode = 0
+    if param0 is None:
+        param0 = 0
+    if param1 is None:
+        param1 = 0
+
+    opcode &= 0xFFFF
+    param0 &= 0xFFFF
+    param1 &= 0xFFFF
+    crc_word = opcode ^ param0 ^ param1
+
+    data = flipfour(AD9102_WAVE_CTRL_HEADER)  # Word 0 (header)
+    data += flipfour(int_to_hex(opcode))
+    data += flipfour(int_to_hex(param0))
+    data += flipfour(int_to_hex(param1))
+    data += flipfour(int_to_hex(crc_word))
+
+    return bytearray.fromhex(data)
+
+
+def create_AD9102_waveform_data_command(samples_chunk):
+    if samples_chunk is None:
+        raise ValueError("AD9102 waveform chunk is missing.")
+
+    samples = list(samples_chunk)
+    chunk_count = len(samples)
+    if chunk_count < 1 or chunk_count > AD9102_WAVE_CHUNK_SAMPLES:
+        raise ValueError("AD9102 waveform chunk size must be within [1, 12].")
+
+    encoded_words = [chunk_count]
+    for sample in samples:
+        if isinstance(sample, bool) or not isinstance(sample, int):
+            raise ValueError("AD9102 waveform samples must be integers.")
+        if sample < AD9102_SAMPLE_MIN or sample > AD9102_SAMPLE_MAX:
+            raise ValueError("AD9102 waveform sample is out of range [-8192, 8191].")
+        encoded_words.append(sample & 0xFFFF)
+
+    while len(encoded_words) < (1 + AD9102_WAVE_CHUNK_SAMPLES):
+        encoded_words.append(0)
+
+    crc_word = 0
+    for word in encoded_words:
+        crc_word ^= word
+
+    data = flipfour(AD9102_WAVE_DATA_HEADER)  # Word 0 (header)
+    data += flipfour(int_to_hex(encoded_words[0]))
+    for word in encoded_words[1:]:
+        data += flipfour(int_to_hex(word))
+    data += flipfour(int_to_hex(crc_word))
+
+    return bytearray.fromhex(data)
+
+
 def encode_Input(params):
 
     if params is None: