diff --git a/spoolbuddy/scripts/read_tag.py b/spoolbuddy/scripts/read_tag.py index b8e95a65d..ffcdbf51c 100644 --- a/spoolbuddy/scripts/read_tag.py +++ b/spoolbuddy/scripts/read_tag.py @@ -1,23 +1,550 @@ #!/usr/bin/env python3 -"""PN5180 NFC Tag Reader diagnostic script. +"""PN5180 NFC tag reader — ported from working Pico firmware (pico-nfc-bridge.ino). -Standalone diagnostic — the PN5180 driver lives in -spoolbuddy/daemon/pn5180.py and is imported from there. - -Supports: Bambu (MIFARE Classic) + NTAG (SpoolEase/OpenPrintTag) +Key learnings from pico-nfc-bridge.ino: +- Must call setTransceiveMode() before every SEND_DATA +- waitBusy() must wait for HIGH then LOW (not just LOW) +- Bambu tags are MIFARE Classic 1K (ISO 14443A), not ISO 15693 +- SPI at 500kHz, 5µs CS setup, 100µs post-CS delay +- MFC_AUTHENTICATE (0x0C) is a PN5180 host command — Crypto1 handled in hardware +- HKDF-SHA256 derives per-sector keys from master key + UID """ -import logging +import hashlib +import hmac +import os import sys import time -from pathlib import Path -# Add daemon package to sys.path so we can import the driver -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from daemon.pn5180 import BAMBU_BLOCKS, PN5180 +import gpiod +import spidev -# Show driver debug output during diagnostics -logging.basicConfig(level=logging.DEBUG, format=" %(message)s") + +def _env_int(name: str, default: int) -> int: + value = os.environ.get(name) + if value is None or value == "": + return default + try: + return int(value) + except ValueError: + return default + + +BUSY_PIN = _env_int("SPOOLBUDDY_NFC_BUSY_PIN", 25) +RST_PIN = _env_int("SPOOLBUDDY_NFC_RST_PIN", 24) +NSS_PIN = _env_int("SPOOLBUDDY_NFC_NSS_PIN", 23) # Manual CS by default +SPI_BUS = _env_int("SPOOLBUDDY_NFC_SPI_BUS", 0) +SPI_DEVICE = _env_int("SPOOLBUDDY_NFC_SPI_DEVICE", 0) +SPI_SPEED_HZ = _env_int("SPOOLBUDDY_NFC_SPI_SPEED_HZ", 500_000) + +# Bambu Lab MIFARE Classic key derivation constants (from pico-nfc-bridge.ino) +BAMBU_MASTER_KEY = bytes( + [ + 0x9A, + 0x75, + 0x9C, + 0xF2, + 0xC4, + 0xF7, + 0xCA, + 0xFF, + 0x22, + 0x2C, + 0xB9, + 0x76, + 0x9B, + 0x41, + 0xBC, + 0x96, + ] +) +BAMBU_CONTEXT = b"RFID-A\x00" # 7 bytes including null terminator + +# Blocks to read for Bambu tag data +BAMBU_BLOCKS = [1, 2, 4, 5] + + +def hkdf_derive_keys(uid: bytes) -> bytes: + """Derive 96 bytes of MIFARE key material (16 sectors * 6 bytes each). + + Uses HKDF-SHA256 with the Bambu master key as salt and the tag UID as IKM. + """ + # HKDF-Extract: PRK = HMAC-SHA256(salt=master_key, IKM=uid) + prk = hmac.new(BAMBU_MASTER_KEY, uid, hashlib.sha256).digest() + + # HKDF-Expand: generate 96 bytes using context "RFID-A\0" + okm = b"" + t = b"" + counter = 1 + while len(okm) < 96: + t = hmac.new(prk, t + BAMBU_CONTEXT + bytes([counter]), hashlib.sha256).digest() + okm += t + counter += 1 + return okm[:96] + + +def get_sector_key(keys: bytes, block: int) -> bytes: + """Get the 6-byte key for the sector containing the given block.""" + sector = block // 4 + return keys[sector * 6 : sector * 6 + 6] + + +def _find_gpio_chip(): + for path in ["/dev/gpiochip4", "/dev/gpiochip0"]: + try: + chip = gpiod.Chip(path) + if "pinctrl" in chip.get_info().label: + return chip + chip.close() + except (FileNotFoundError, PermissionError, OSError): + continue + raise RuntimeError("No GPIO chip") + + +class PN5180: + def __init__(self): + self._chip = _find_gpio_chip() + self._lines = self._chip.request_lines( + consumer="pn5180", + config={ + BUSY_PIN: gpiod.LineSettings(direction=gpiod.line.Direction.INPUT), + RST_PIN: gpiod.LineSettings( + direction=gpiod.line.Direction.OUTPUT, output_value=gpiod.line.Value.ACTIVE + ), + NSS_PIN: gpiod.LineSettings( + direction=gpiod.line.Direction.OUTPUT, output_value=gpiod.line.Value.ACTIVE + ), + }, + ) + self._spi = spidev.SpiDev() + self._spi.open(SPI_BUS, SPI_DEVICE) + self._spi.max_speed_hz = SPI_SPEED_HZ + self._spi.mode = 0b00 + self._spi.no_cs = True + + def close(self): + self._spi.close() + self._lines.release() + self._chip.close() + + def _cs_low(self): + self._lines.set_value(NSS_PIN, gpiod.line.Value.INACTIVE) + time.sleep(0.000005) # 5µs setup + + def _cs_high(self): + self._lines.set_value(NSS_PIN, gpiod.line.Value.ACTIVE) + time.sleep(0.000100) # 100µs post-CS delay + + def _wait_busy(self, timeout_s=1.0): + """Wait for BUSY to go HIGH (processing) then LOW (done) — matches Pico firmware.""" + deadline = time.monotonic() + min(timeout_s, 0.010) + # Wait for BUSY HIGH (PN5180 started processing) + while self._lines.get_value(BUSY_PIN) != gpiod.line.Value.ACTIVE: + if time.monotonic() > deadline: + break # Timeout waiting for HIGH — command may have processed already + time.sleep(0.00001) + # Wait for BUSY LOW (PN5180 done) + deadline = time.monotonic() + timeout_s + while self._lines.get_value(BUSY_PIN) == gpiod.line.Value.ACTIVE: + if time.monotonic() > deadline: + raise TimeoutError("BUSY timeout") + time.sleep(0.0001) + + def _cmd(self, data): + self._cs_low() + self._spi.xfer2(list(data)) + self._cs_high() + self._wait_busy() + + def _read_response(self, n): + self._cs_low() + result = self._spi.xfer2([0xFF] * n) + self._cs_high() + return result + + # -- Register ops -- + + def write_reg(self, reg, val): + self._cmd([0x00, reg, val & 0xFF, (val >> 8) & 0xFF, (val >> 16) & 0xFF, (val >> 24) & 0xFF]) + + def write_reg_or(self, reg, mask): + self._cmd([0x01, reg, mask & 0xFF, (mask >> 8) & 0xFF, (mask >> 16) & 0xFF, (mask >> 24) & 0xFF]) + + def write_reg_and(self, reg, mask): + self._cmd([0x02, reg, mask & 0xFF, (mask >> 8) & 0xFF, (mask >> 16) & 0xFF, (mask >> 24) & 0xFF]) + + def read_reg(self, reg): + self._cmd([0x04, reg]) + time.sleep(0.000100) # Extra 100µs before read + return int.from_bytes(self._read_response(4), "little") + + def read_eeprom(self, addr, length): + self._cmd([0x07, addr, length]) + time.sleep(0.000100) + return bytes(self._read_response(length)) + + # -- Commands -- + + def reset(self): + self._lines.set_value(RST_PIN, gpiod.line.Value.INACTIVE) + time.sleep(0.050) + self._lines.set_value(RST_PIN, gpiod.line.Value.ACTIVE) + time.sleep(0.100) + self._wait_busy(2.0) + time.sleep(0.050) + + def load_rf_config(self, tx, rx): + self.write_reg(0x03, 0xFFFFFFFF) # Clear IRQs first + time.sleep(0.000100) + self._cmd([0x11, tx, rx]) + time.sleep(0.010) + + def rf_on(self): + self._cmd([0x16, 0x00]) + time.sleep(0.010) + + def rf_off(self): + self._cmd([0x17, 0x00]) + time.sleep(0.005) + + def set_transceive_mode(self): + """Set SYSTEM_CONFIG command bits to TRANSCEIVE (0x03) — CRITICAL!""" + sys_cfg = self.read_reg(0x00) + sys_cfg = (sys_cfg & 0xFFFFFFF8) | 0x03 + self.write_reg(0x00, sys_cfg) + + def send_data(self, data, valid_bits=0x00): + self._cs_low() + self._spi.xfer2([0x09, valid_bits] + list(data)) + self._cs_high() + time.sleep(0.000100) + self._wait_busy() + + def read_data(self, length): + self._cmd([0x0A, 0x00]) + return bytes(self._read_response(length)) + + # -- ISO 14443A -- + + def activate_type_a(self): + """Full Type A activation: WUPA -> Anticollision -> SELECT. Returns (uid, sak) or None.""" + # Crypto off, CRC off + self.write_reg_and(0x00, 0xFFFFFFBF) + self.write_reg_and(0x12, 0xFFFFFFFE) + self.write_reg_and(0x19, 0xFFFFFFFE) + self.write_reg(0x03, 0xFFFFFFFF) + + # Reset to IDLE then TRANSCEIVE + sys_cfg = self.read_reg(0x00) + self.write_reg(0x00, sys_cfg & 0xFFFFFFF8) # IDLE + time.sleep(0.001) + self.write_reg(0x00, (sys_cfg & 0xFFFFFFF8) | 0x03) # TRANSCEIVE + time.sleep(0.002) + + # WUPA (7-bit) + self.send_data([0x52], valid_bits=0x07) + time.sleep(0.005) + + rx_status = self.read_reg(0x13) + rx_len = rx_status & 0x1FF + if rx_len < 2 or rx_len == 511: + # Try REQA + self.write_reg(0x03, 0xFFFFFFFF) + time.sleep(0.002) + self.set_transceive_mode() + time.sleep(0.002) + self.send_data([0x26], valid_bits=0x07) + time.sleep(0.005) + rx_status = self.read_reg(0x13) + rx_len = rx_status & 0x1FF + if rx_len < 2 or rx_len == 511: + return None + + atqa = self.read_data(2) + if atqa[0] == 0xFF or atqa[0] == 0x00: + return None + + # Anti-collision Level 1 + self.write_reg(0x03, 0xFFFFFFFF) + self.set_transceive_mode() + time.sleep(0.002) + + self.send_data([0x93, 0x20]) + time.sleep(0.010) + + rx_status = self.read_reg(0x13) + rx_len = rx_status & 0x1FF + if rx_len < 5 or rx_len > 64: + return None + + uid_buf = self.read_data(5) + uid = uid_buf[:4] + bcc = uid[0] ^ uid[1] ^ uid[2] ^ uid[3] + if bcc != uid_buf[4]: + return None + + # SELECT + self.write_reg(0x03, 0xFFFFFFFF) + self.set_transceive_mode() + time.sleep(0.002) + + # Enable CRC for SELECT + self.write_reg_or(0x19, 0x01) + self.write_reg_or(0x12, 0x01) + + self.send_data([0x93, 0x70, uid[0], uid[1], uid[2], uid[3], bcc]) + time.sleep(0.010) + + rx_status = self.read_reg(0x13) + rx_len = rx_status & 0x1FF + if rx_len < 1: + return None + + sak_buf = self.read_data(min(rx_len, 3)) + sak = sak_buf[0] + + return bytes(uid), sak + + # -- MIFARE Classic -- + + def mfc_authenticate(self, block: int, key: bytes, uid: bytes) -> bool: + """MIFARE Classic authentication via PN5180 MFC_AUTHENTICATE (0x0C). + + The PN5180 handles Crypto1 internally. After success, bit 6 of + SYSTEM_CONFIG is set (MFC_CRYPTO1_ON) and all subsequent RF + communication is encrypted/decrypted by the hardware. + + Args: + block: Block number to authenticate + key: 6-byte MIFARE Key A + uid: 4-byte tag UID + Returns: + True if authentication succeeded + """ + # Wait for BUSY LOW before starting + deadline = time.monotonic() + 0.100 + while self._lines.get_value(BUSY_PIN) == gpiod.line.Value.ACTIVE: + if time.monotonic() > deadline: + return False + time.sleep(0.001) + + # MFC_AUTHENTICATE: [0x0C][key 6B][keyType][blockNo][uid 4B] = 13 bytes + cmd = [0x0C] + list(key) + [0x60, block] + list(uid[:4]) + self._cs_low() + self._spi.xfer2(cmd) + self._cs_high() + + # Wait for BUSY HIGH then LOW (auth can take up to 1s) + self._wait_busy(timeout_s=1.0) + + # Read 1-byte response: 0x00 = success + self._cs_low() + response = self._spi.xfer2([0xFF]) + self._cs_high() + + return response[0] == 0x00 + + def mfc_read_block(self, block: int) -> bytes | None: + """Read a 16-byte MIFARE Classic block (must be authenticated first). + + Returns 16 bytes of block data, or None on failure. + """ + # Clear IRQs + self.write_reg(0x03, 0xFFFFFFFF) + + # Set transceive mode (Crypto1 stays active from MFC_AUTHENTICATE) + self.set_transceive_mode() + time.sleep(0.001) + + # Enable TX and RX CRC for encrypted read + self.write_reg_or(0x19, 0x01) + self.write_reg_or(0x12, 0x01) + + # Send MIFARE READ command: 0x30 + block number + self.send_data([0x30, block]) + time.sleep(0.010) + + # Check RX status + rx_status = self.read_reg(0x13) + rx_len = rx_status & 0x1FF + if rx_len != 16: + return None + + return self.read_data(16) + + def ntag_read_pages(self, start_page: int, num_pages: int) -> bytes | None: + """Read NTAG pages (4 bytes each). No authentication required. + + Uses NTAG READ command (0x30) which returns 4 pages (16 bytes) at a time. + CRC must be disabled for NTAG reads. + """ + # Disable CRC for NTAG + self.write_reg_and(0x19, 0xFFFFFFFE) # TX CRC off + self.write_reg_and(0x12, 0xFFFFFFFE) # RX CRC off + + result = bytearray() + pages_read = 0 + while pages_read < num_pages: + self.write_reg(0x03, 0xFFFFFFFF) # Clear IRQs + self.set_transceive_mode() + time.sleep(0.001) + + # READ command: 0x30 + page number → returns 16 bytes (4 pages) + self.send_data([0x30, start_page + pages_read]) + time.sleep(0.005) + + rx_status = self.read_reg(0x13) + rx_len = rx_status & 0x1FF + if rx_len < 16: + return None + + data = self.read_data(16) + # Copy only the pages we need + pages_to_copy = min(4, num_pages - pages_read) + result.extend(data[: pages_to_copy * 4]) + pages_read += 4 # Always advances by 4 (READ returns 4 pages) + + return bytes(result) + + def reactivate_card(self) -> tuple[bytes, int] | None: + """RF cycle and full re-select of the card. Returns (uid, sak) or None.""" + self.rf_off() + time.sleep(0.010) + + self.write_reg(0x03, 0xFFFFFFFF) # Clear IRQs + self.load_rf_config(0x00, 0x80) # ISO 14443A + time.sleep(0.005) + + self.rf_on() + time.sleep(0.020) + + return self.activate_type_a() + + def read_bambu_tag(self, uid: bytes) -> dict[int, bytes] | None: + """Read Bambu tag data blocks using HKDF-derived keys. + + Args: + uid: 4-byte tag UID (from activate_type_a) + Returns: + Dict mapping block number -> 16 bytes of data, or None on failure + """ + # Derive per-sector keys from UID + keys = hkdf_derive_keys(uid) + + # Clear Crypto1 state and IRQs + self.write_reg_and(0x00, 0xFFFFFFBF) # Clear MFC_CRYPTO1_ON (bit 6) + self.write_reg(0x03, 0xFFFFFFFF) + + # Reactivate card (may have timed out) + result = self.reactivate_card() + if result is None: + print(" Failed to reactivate card") + return None + + uid_check, _ = result + if uid_check != uid: + print(f" UID mismatch after reactivation: {uid_check.hex()} != {uid.hex()}") + return None + + # Read blocks with per-sector authentication + blocks = {} + current_sector = -1 + + for block in BAMBU_BLOCKS: + sector = block // 4 + + # Authenticate when entering a new sector + if sector != current_sector: + key = get_sector_key(keys, block) + if not self.mfc_authenticate(block, key, uid): + print(f" Auth failed for block {block} (sector {sector})") + return None + current_sector = sector + + # Read the block + data = self.mfc_read_block(block) + if data is None: + print(f" Read failed for block {block}") + return None + blocks[block] = data + + return blocks + + def ntag_write_page(self, page: int, data: bytes) -> bool: + """Write 4 bytes to a single NTAG page. + + NTAG WRITE command: 0xA2 + page_number + 4 bytes data. + CRC disabled (same as reads). Returns True on ACK (0x0A). + """ + if len(data) != 4: + return False + + # Disable CRC + self.write_reg_and(0x19, 0xFFFFFFFE) # TX CRC off + self.write_reg_and(0x12, 0xFFFFFFFE) # RX CRC off + + # Clear IRQs and set transceive mode + self.write_reg(0x03, 0xFFFFFFFF) + self.set_transceive_mode() + time.sleep(0.001) + + # WRITE command: 0xA2 + page + 4 bytes + self.send_data([0xA2, page] + list(data)) + time.sleep(0.005) + + # Check for ACK: NTAG ACK is 4-bit 0x0A + rx_status = self.read_reg(0x13) + rx_len = rx_status & 0x1FF + if rx_len < 1: + return False + + ack = self.read_data(1) + return ack[0] == 0x0A + + def ntag_write_pages(self, start_page: int, data: bytes) -> bool: + """Write data to consecutive NTAG pages starting at start_page. + + Pads last chunk to 4 bytes. Verifies by reading back. + Returns True if write + verify succeeded. + """ + # Pad to 4-byte boundary + padded = bytearray(data) + while len(padded) % 4 != 0: + padded.append(0x00) + + # Write page by page + for i in range(0, len(padded), 4): + page = start_page + (i // 4) + chunk = bytes(padded[i : i + 4]) + if not self.ntag_write_page(page, chunk): + return False + time.sleep(0.002) + + # Reactivate card for verification read + result = self.reactivate_card() + if result is None: + return False + + # Read back and verify + num_pages = len(padded) // 4 + readback = self.ntag_read_pages(start_page, num_pages) + if readback is None: + return False + + return readback[: len(data)] == data + + def read_ntag(self, uid: bytes) -> bytes | None: + """Read NTAG pages 4-20 (NDEF data area, 68 bytes). No auth needed. + + Used for SpoolEase / OpenPrintTag community tags. + """ + # Reactivate card + result = self.reactivate_card() + if result is None: + print(" Failed to reactivate card") + return None + + return self.ntag_read_pages(start_page=4, num_pages=17) def _print_hex_dump(data: bytes, label: str, bytes_per_line: int = 16): @@ -56,7 +583,7 @@ def main(): else: print("\nCheck:") print(" - Correct GPIO chip is available (/dev/gpiochip0 or /dev/gpiochip4)") - print(" - SPI device is available") + print(f" - SPI device is available (SPI_BUS={SPI_BUS}, SPI_DEVICE={SPI_DEVICE})") print(" - GPIO and SPI permissions are correct") # Only print full traceback for unexpected errors import traceback diff --git a/spoolbuddy/scripts/scale_diag.py b/spoolbuddy/scripts/scale_diag.py index a557cb937..2759b5472 100644 --- a/spoolbuddy/scripts/scale_diag.py +++ b/spoolbuddy/scripts/scale_diag.py @@ -1,23 +1,152 @@ #!/usr/bin/env python3 -"""NAU7802 Scale Diagnostic. +"""NAU7802 Scale Diagnostic - ported from SpoolBuddy Rust firmware. -Standalone diagnostic script — the NAU7802 driver lives in -spoolbuddy/daemon/nau7802.py and is imported from there. +I2C address: 0x2A +Bus: /dev/i2c-1 (GPIO2/GPIO3 on RPi) """ -import logging +import os +import struct import sys import time -from pathlib import Path import smbus2 -# Add daemon package to sys.path so we can import the driver -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from daemon.nau7802 import I2C_BUS, NAU7802, NAU7802_ADDR -# Show driver debug output (init steps, etc.) during diagnostics -logging.basicConfig(level=logging.DEBUG, format=" %(message)s") +def _env_int(name: str, default: int) -> int: + value = os.environ.get(name) + if value is None or value == "": + return default + try: + return int(value) + except ValueError: + return default + + +I2C_BUS = _env_int("SPOOLBUDDY_I2C_BUS", 1) +NAU7802_ADDR = 0x2A + +# Register addresses +REG_PU_CTRL = 0x00 +REG_CTRL1 = 0x01 +REG_CTRL2 = 0x02 +REG_ADCO_B2 = 0x12 # ADC output MSB +REG_ADCO_B1 = 0x13 +REG_ADCO_B0 = 0x14 # ADC output LSB +REG_ADC = 0x15 +REG_PGA = 0x1B +REG_PWR_CTRL = 0x1C +REG_REVISION = 0x1F + +# PU_CTRL bits +PU_RR = 0x01 # Register reset +PU_PUD = 0x02 # Power up digital +PU_PUA = 0x04 # Power up analog +PU_PUR = 0x08 # Power up ready (read-only) +PU_CS = 0x10 # Cycle start +PU_CR = 0x20 # Cycle ready (read-only) +PU_OSCS = 0x40 # Oscillator select +PU_AVDDS = 0x80 # AVDD source select + + +class NAU7802: + def __init__(self, bus=I2C_BUS, addr=NAU7802_ADDR): + self._bus_num = bus + self._bus = smbus2.SMBus(bus) + self._addr = addr + + def close(self): + self._bus.close() + + def read_reg(self, reg: int) -> int: + return self._bus.read_byte_data(self._addr, reg) + + def write_reg(self, reg: int, val: int): + self._bus.write_byte_data(self._addr, reg, val & 0xFF) + + def _update_bits(self, reg: int, mask: int, value: int): + cur = self.read_reg(reg) + self.write_reg(reg, (cur & ~mask) | (value & mask)) + + def _set_bit(self, reg: int, bit: int, enabled: bool): + mask = 1 << bit + self._update_bits(reg, mask, mask if enabled else 0) + + def _set_field(self, reg: int, shift: int, width: int, value: int): + mask = ((1 << width) - 1) << shift + self._update_bits(reg, mask, value << shift) + + def init(self): + """Initialize NAU7802 using the Adafruit library startup sequence.""" + + # Reset + self._set_bit(REG_PU_CTRL, 0, True) # RR=1 + time.sleep(0.010) + self._set_bit(REG_PU_CTRL, 0, False) # RR=0 + self._set_bit(REG_PU_CTRL, 1, True) # PUD=1 + time.sleep(0.001) + + # Enable digital + analog and allow analog section to settle. + self._set_bit(REG_PU_CTRL, 1, True) # PUD=1 + self._set_bit(REG_PU_CTRL, 2, True) # PUA=1 + time.sleep(0.600) + + # Start conversion cycle (PU_CS bit 4) after power-up. + self._set_bit(REG_PU_CTRL, 4, True) + + # Wait for power-up ready (PU_PUR bit 3) + for _ in range(100): + status = self.read_reg(REG_PU_CTRL) + if status & PU_PUR: + print(" Power-up ready") + break + time.sleep(0.001) + else: + raise TimeoutError("NAU7802 power-up timeout") + + # Check revision register low nibble (Adafruit expects 0xF). + revision = self.read_reg(REG_REVISION) + print(f" Revision: 0x{revision:02X}") + if (revision & 0x0F) != 0x0F: + raise RuntimeError(f"Unexpected NAU7802 revision register: 0x{revision:02X}") + + # Internal LDO enable is PU_CTRL.AVDDS (bit 7); set LDO voltage to 3.0V. + self._set_bit(REG_PU_CTRL, 7, True) # AVDDS=1 (internal LDO) + self._set_field(REG_CTRL1, shift=3, width=3, value=0b101) # VLDO=3.0V + print(" LDO: 3.0V (internal)") + + # Gain: 128x (bits 2:0 of CTRL1 = 0b111) + self._set_field(REG_CTRL1, shift=0, width=3, value=0b111) + print(" Gain: 128x") + + # Sample rate: 10 SPS (CTRL2 bits 6:4 = 0b000) + self._set_field(REG_CTRL2, shift=4, width=3, value=0b000) + print(" Sample rate: 10 SPS") + + # Adafruit tuning: disable ADC chopper clock (ADC bits 5:4 = 0b11) + self._set_field(REG_ADC, shift=4, width=2, value=0b11) + + # Adafruit tuning: use low ESR caps (PGA bit 6 = 0) + self._set_bit(REG_PGA, 6, False) + + # Start conversion cycle + self._set_bit(REG_PU_CTRL, 4, True) + print(" Conversion started") + + def data_ready(self) -> bool: + return bool(self.read_reg(REG_PU_CTRL) & PU_CR) + + def read_raw(self) -> int: + """Read 24-bit signed ADC value.""" + b2 = self.read_reg(REG_ADCO_B2) + b1 = self.read_reg(REG_ADCO_B1) + b0 = self.read_reg(REG_ADCO_B0) + raw = (b2 << 16) | (b1 << 8) | b0 + # Sign extend 24-bit to 32-bit + if raw & 0x800000: + raw |= 0xFF000000 + raw = struct.unpack("i", struct.pack("I", raw))[0] + return raw def main(): @@ -62,7 +191,6 @@ def main(): try: print("[1] Initializing...") scale.init() - print(" Initialized OK") print("[2] Waiting for first reading...") for _ in range(200):