#!/usr/bin/env python3 """ ================================================== Software Version: 20260429.1200 ================================================== """ import time import spidev import pigpio from adc_commands import ADCCommand, OSR, OSR_REGISTER_MAP from typing import Union, List, Optional, Tuple class ADS131M04: # Programmable Gain Amplifier mapping: index → actual gain multiplier GAIN_VALS = [1, 2, 4, 8, 16, 32, 64, 128] # Register addresses REG_GAIN = 0x04 # GAIN REGISTER REG_MODE = 0x02 # MODE REGISTER REG_CLOCK = 0x03 # CLOCK REGISTER (contains OSR, channel enables, PWR) def __init__(self, pi, spi_bus: int = 0, spi_dev: int = 0, baud: int = 8000000, drdy_pin: int = 22, sync_pin: int = 24, osr: int = 1024): """ Initialize ADS131M04 driver instance. Parameters: • pi - pigpio instance (for GPIO control) • spi_bus - SPI bus number (default: 0) • spi_dev - SPI device number (default: 0) • baud - SPI baud rate in Hz (default: 8MHz) • drdy_pin - GPIO pin for DRDY signal (default: 23) • sync_pin - GPIO pin for SYNC/RESET signal (default: 25) • osr - Oversampling ratio (default: 128) """ # Store configuration parameters self.pi = pi self.drdy_pin = drdy_pin self.sync_pin = sync_pin self.osr = osr # Word size configuration (ADS131M04 defaults to 24-bit on power-up) self.word_length = 24 # bits self.bytes_per_word = 3 # 24 bits = 3 bytes # Initialize SPI interface self.spi = spidev.SpiDev() self.spi.open(spi_bus, spi_dev) self.spi.max_speed_hz = baud self.spi.mode = 0b01 # SPI Mode 1 (CPOL=0, CPHA=1) self.spi.no_cs = True def init(self): """ Initialize and configure the ADS131M04 ADC. """ # Verify pigpio connection if not self.pi.connected: raise RuntimeError("pigpio daemon not connected") # Setup GPIO pins self.pi.set_mode(self.drdy_pin, pigpio.INPUT) self.pi.set_mode(self.sync_pin, pigpio.OUTPUT) time.sleep(1) # Reset ADC and verify communication self.pi.write(self.sync_pin, 1) # deactivate SYNC self.sync_pulse() rx = self.spi.xfer2([0x00] * 15) print("nulll response:", [hex(b) for b in rx]) mode_read = self.read_register(self.REG_CLOCK) print("reg REG_CLOCK =", hex(mode_read)) time.sleep(0.5) mode_val = 0x0510 self.write_register(self.REG_MODE, mode_val) time.sleep(0.001) mode_read = self.read_register(self.REG_MODE) print("MODE reg after write =", hex(mode_read)) mode_read = self.read_register(self.REG_MODE) print("reg REG_MODE =", hex(mode_read)) mode_read = self.read_register(self.REG_CLOCK) print("reg REG_CLOCK =", hex(mode_read)) mode_read = self.read_register(0x00) print("reg ID =", hex(mode_read)) def read_register(self, reg_addr: int, num_regs: int = 1): WORD_SIZE = 3 # 24-bit words FRAME_BYTES = 15 # 5 words × 3 bytes # --------------------------------------- # 1. Build RREG command # --------------------------------------- cmd_word = (0b101 << 13) | (reg_addr << 7) | (num_regs - 1) cmd_high = (cmd_word >> 8) & 0xFF cmd_low = cmd_word & 0xFF tx_cmd = [cmd_high, cmd_low, 0x00] + [0x00] * (FRAME_BYTES - 3) # --------------------------------------- # 2. FRAME 1 → send RREG command # --------------------------------------- self.spi.xfer2(tx_cmd) # --------------------------------------- # 3. FRAME 2 → read response # --------------------------------------- rx = self.spi.xfer2([0x00] * FRAME_BYTES) print("RREG RX:", [hex(b) for b in rx]) # --------------------------------------- # 4. Parse response # --------------------------------------- # SINGLE REGISTER if num_regs == 1: # First word = REGISTER DATA return (rx[0] << 8) | rx[1] # MULTIPLE REGISTERS reg_values = [] for i in range(num_regs): offset = (i) * WORD_SIZE val = (rx[offset] << 8) | rx[offset + 1] reg_values.append(val) return reg_values def write_register(self, reg_addr: int, reg_values): WORD_SIZE = 3 # 24-bit words FRAME_WORDS = 5 # fixed frame size FRAME_BYTES = FRAME_WORDS * WORD_SIZE # 15 bytes # --------------------------------------- # 1. Prepare register values # --------------------------------------- if isinstance(reg_values, int): values_list = [reg_values] else: values_list = reg_values num_regs = len(values_list) # --------------------------------------- # 2. Build WREG command word (16-bit) # Format: 011 aaaaa a nnn nnnn # --------------------------------------- cmd_word = (0b011 << 13) | (reg_addr << 7) | (num_regs - 1) cmd_high = (cmd_word >> 8) & 0xFF cmd_low = cmd_word & 0xFF # --------------------------------------- # 3. Build full SPI frame (5 words) # Word 0 → COMMAND # Word 1..N → REGISTER DATA # Remaining words → ZERO padding # --------------------------------------- tx_frame = [] # Word 0 → COMMAND tx_frame.extend([cmd_high, cmd_low, 0x00]) # --------------------------------------- # 4. Add register data words # Each register value = 16-bit MSB aligned # Stored as: [MSB][LSB][PAD] # --------------------------------------- for val in values_list: tx_frame.extend([ (val >> 8) & 0xFF, # MSB val & 0xFF, # LSB 0x00 # PAD (for 24-bit alignment) ]) # --------------------------------------- # 5. Pad remaining words to complete frame # Full frame must be exactly 5 words # --------------------------------------- while len(tx_frame) < FRAME_BYTES: tx_frame.extend([0x00, 0x00, 0x00]) print("WREG TX:", [hex(x) for x in tx_frame]) # --------------------------------------- # 6. FRAME 1 → Send WREG command + data # --------------------------------------- self.spi.xfer2(tx_frame) # --------------------------------------- # 7. FRAME 2 → Read WREG response (ACK) # First word = WREG response (010 pattern) # --------------------------------------- rx = self.spi.xfer2([0x00] * FRAME_BYTES) print("WREG RX:", [hex(x) for x in rx]) # --------------------------------------- # 8. Parse ACK (first word) # --------------------------------------- ack = (rx[0] << 8) | rx[1] print(f"WREG ACK: 0x{ack:04X}") return ack def set_gain(self, channel: Union[int, List[int]], gain: Union[int, List[int]]): """ Set the programmable gain amplifier (PGA) for one or more channels. Parameters: • channel - Single channel (0-3) or list of channels • gain - Single gain value or list of gain values (must be one of: 1, 2, 4, 8, 16, 32, 64, 128) Examples: • set_gain(0, 16) - Set channel 0 to gain 16 • set_gain([0, 1, 2, 3], 16) - Set all channels to gain 16 • set_gain([0, 1], [16, 32]) - Set channel 0 to gain 16, channel 1 to gain 32 """ if isinstance(channel, int): channel_list = [channel] # convert single value to list else: channel_list = channel if isinstance(gain, int): gain_list = [gain] * len(channel_list) # Same gain for all channels else: gain_list = gain if len(channel_list) != len(gain_list): # verify channel gains = channel count raise ValueError(f"Number of channels ({len(channel_list)}) must match number of gains ({len(gain_list)})") gain_value = [] for g in gain_list: # convert gain indicies to gain values if g not in self.GAIN_VALS: raise ValueError(f"Invalid gain value {g}. Must be one of {self.GAIN_VALS}") gain_value.append(self.GAIN_VALS.index(g)) # Read current GAIN register - all 4 channel gains. current_gain_reg = self.read_register(self.REG_GAIN) # Modify gain bits for specified channels new_gain_reg = current_gain_reg for ch, gain_val in zip(channel_list, gain_value): if ch < 0 or ch > 3: raise ValueError(f"Invalid channel {ch}. Must be 0-3") shift = ch * 4 # Calculate bit position for this channel mask = 0x7 << shift # 0x7 = 0b111. 3-bit mask for correct position new_gain_reg = (new_gain_reg & ~mask) | (gain_val << shift) # Clear old gain bits and set new gain value self.write_register(self.REG_GAIN, new_gain_reg) def set_osr(self, osr: int): """ Set the oversampling ratio (OSR) for the ADC. OSR bits are in CLOCK register (0x03), bits [4:2]. Parameters: • osr - Oversampling ratio (use OSR constants from adc_commands) """ if osr not in OSR_REGISTER_MAP: valid_osr = list(OSR_REGISTER_MAP.keys()) raise ValueError(f"Invalid OSR: {osr}. Must be one of {valid_osr}") clock_val = 0x0F00 | (OSR_REGISTER_MAP[osr] << 2) | 0x02 print(f"set_osr: 0x{clock_val:04X}") self.write_register(self.REG_CLOCK,clock_val) self.osr = osr def sync_pulse(self): """ Toggle the SYNC/RESET pin before starting data collection to reset and synchronize the ADC for a clean start. (Active Low) """ self.pi.write(self.sync_pin, 0) # activate SYNC time.sleep(0.000500) # 300 microseconds LOW pulse self.pi.write(self.sync_pin, 1) # deactivate SYNC time.sleep(0.001250) # 150 microseconds wait for ADC to stabilize def read_sample(self, timeout_ms: int = 10) -> Optional[Tuple[int, int, int, int, int]]: """ Read one ADC sample from all 4 channels. Waits for DRDY pin to go LOW, then performs SPI read. Compatible with 24-bit word size (device default). Parameters: • timeout_ms - Maximum time to wait for DRDY in milliseconds (default: 10ms) Return value: • Returns tuple: (status, ch0, ch1, ch2, ch3) - status: 16-bit status register value - ch0-ch3: 24-bit signed ADC values (two's complement) • Returns None if timeout occurs or read fails Frame structure (24-bit mode, 15 bytes total): [STATUS:3B][CH0:3B][CH1:3B][CH2:3B][CH3:3B] """ start_time = time.time() # current time in seconds timeout_sec = timeout_ms / 1000.0 while True: drdy_state = self.pi.read(self.drdy_pin) # read DRDY status if drdy_state == 0: # DRDY is Low break if (time.time() - start_time) > timeout_sec: return None time.sleep(0.000001) null_cmd = [0x00, 0x00, 0x00] frame_size = 5 * self.bytes_per_word # 5 words × 3 bytes = 15 bytes tx_data = null_cmd + ([0x00] * (frame_size - len(null_cmd))) rx_data = self.spi.xfer2(tx_data) # Send 15 null bytes to trigger SPI CLK if len(rx_data) != frame_size: return None # Invalid frame size # Verify we received data if rx_data is None or len(rx_data) == 0: return None # No data received status = (rx_data[0] << 8) | rx_data[1] # status [MSB][LSB][padding] channels = [] for i in range(4): byte_offset = 3 + (i *3) # to skip 3 status bytes raw_value = (rx_data[byte_offset] << 16) | (rx_data[byte_offset + 1] << 8) | rx_data[byte_offset + 2] # Convert from 24-bit two's complement to signed integer if raw_value & 0x800000: signed_value = raw_value - 0x1000000 # Subtract 2^24 else: signed_value = raw_value Voltage = (signed_value / 2**23) * (1.2 / 1) channels.append(Voltage) # Return tuple: (status, ch0, ch1, ch2, ch3) return (status, channels[0], channels[1], channels[2], channels[3])