#!/usr/bin/env python """ CAN Bus IoT Library for Raspberry Pi - v2.2 Support for digital and analog vertebrae communication via CAN bus. Novetats respecte v2.1: - set_async_adc(addr, side, enable, period_ms): activa/desactiva l'enviament periodic dels valors ADC d'una vertebra analogica (registre ASYNC_CFG, cal firmware >= v1.5). Evita haver de fer polling amb ain(). - on_ain_change(addr, side, callback): listener passiu dels valors ADC enviats periodicament (cal haver activat abans amb set_async_adc()). - off_ain_change(addr, side) / off_all_ain_change(): cancel.len listeners. Novetats respecte v2.0: - on_din_change(addr, callback): listener passiu de canvis digitals espontanis. La vertebra publica canvis sense que ningu els demani (ID 0x420 per addr 0x0). El callback s'executa en un thread de fons; no col.lapsa el bus CAN. - off_din_change(addr): cancel.la el listener d'una adreca. - off_all_din_change(): cancel.la tots els listeners actius. Exemple d'us del listener digital: def my_callback(addr, val_a, val_b): print(f"addr={addr} valA={val_a:#04x} valB={val_b:#04x}") iotv.on_din_change('0000', my_callback) # el programa continua; callback es crida quan hi ha canvis iotv.off_din_change('0000') Exemple d'us de l'asincron analogic: iotv.set_async_adc('0000', 'A', True, 200) # cada 200ms, costat A def my_ain_callback(addr, side, values): print(f"addr={addr} side={side} values={values}") iotv.on_ain_change('0000', 'A', my_ain_callback) # ... iotv.off_ain_change('0000', 'A') iotv.set_async_adc('0000', 'A', False) Usage: pi@raspberrypi:~/codis/canbus $ sudo ip link set can0 type can bitrate 100000 pi@raspberrypi:~/codis/canbus $ sudo ifconfig can0 up pi@raspberrypi:~/codis/canbus $ sudo ifconfig can0 down """ import os import can from datetime import datetime from time import sleep from typing import Optional, Union import RPi.GPIO as GPIO import atexit import threading # ============================================================================= # CONSTANTS # ============================================================================= CAN_CHANNEL = 'can0' CAN_INTERFACE = 'socketcan' CAN_BITRATE = 100000 USE_LEDS = False VERBOSE = False LED_W = 12 LED_B = 13 TIMEOUT_RCV = 0.5 # Arbitration IDs ARB_ID_ANALOG_BASE = 16 ARB_ID_ANALOG_VERSION = 512 ARB_ID_AOUT_BASE = 784 ARB_ID_DIGITAL_BASE = 800 # Registre de configuracio asincrona (vertebra analogica, firmware >= v1.5) ASYNC_CFG = 0x08 # Els missatges ASYNC_CFG (comandes cap a la vertebra) es tramenten al mateix # ID que s'usa per REG_WR (AOUT), ja que la vertebra nomes mira el byte baix # de l'identificador (message.identifier & 0x00FF == byI2cAddr). ARB_ID_ASYNC_CFG_BASE = ARB_ID_AOUT_BASE # Els missatges espontanis de valors ADC (vertebra --> cap) es tramenten a # 0x300 + byI2cAddr + side (side: 0=A, 1=B). Com que ARB_ID_AOUT_BASE=784=0x310 # i byI2cAddr = 0x10 + jumpers, es correspon amb ARB_ID_AOUT_BASE + jumpers + side. ARB_ID_ASYNC_PUSH_BASE = ARB_ID_AOUT_BASE # Voltage conversion constants VOLTAGE_MIN = -10.0 VOLTAGE_MAX = 10.0 AIN_MAX_VALUE = 26624 AOUT_MAX_VALUE = 4095 AOUT_VOLTAGE_MAX = 10.0 # ============================================================================= # GLOBAL BUS INSTANCE (SINGLETON) # ============================================================================= _bus_instance: Optional[can.interface.Bus] = None def get_bus() -> can.interface.Bus: """ Obté la instància global del bus CAN (singleton pattern). Crea la instància si no existeix. Returns: Instància del bus CAN """ global _bus_instance if _bus_instance is None: try: _bus_instance = can.interface.Bus( channel=CAN_CHANNEL, interface=CAN_INTERFACE ) if VERBOSE: print(f"CAN bus initialized on {CAN_CHANNEL}") except can.CanError as e: print(f"Error initializing CAN bus: {e}") raise return _bus_instance def close_bus() -> None: """ Tanca el bus CAN global i allibera els recursos. Atura prèviament tots els listeners de canvis digitals. """ global _bus_instance off_all_din_change() off_all_ain_change() if _bus_instance is not None: try: _bus_instance.shutdown() if VERBOSE: print("CAN bus closed properly") except Exception as e: print(f"Error closing CAN bus: {e}") finally: _bus_instance = None # Registrar el tancament automàtic quan el programa acabi atexit.register(close_bus) # ============================================================================= # DIN CHANGE LISTENER — listener passiu de canvis digitals espontanis # ============================================================================= # La vertebra digital publica missatges espontanis quan detecta un canvi # d'entrada, sense que ningú els demani. Format observat: # arbitration_id = ARB_ID_DIGITAL_BASE + 256 + i2c_addr (ex: 0x420 per addr 0) # data[3] = valA (invertit: ~valA & 0xFF dona el byte de entrades A) # data[4] = valB (invertit: ~valB & 0xFF dona el byte de entrades B) # # ARB_ID_DIGITAL_PUSH_OFFSET: offset per als missatges espontanis vs els RTR ARB_ID_DIGITAL_PUSH_OFFSET = 256 # Diccionari de listeners actius: {i2c_addr: (thread, stop_event, callback)} _din_listeners: dict = {} _din_listeners_lock = threading.Lock() def _din_listener_thread(i2c_addr: int, stop_event: threading.Event, callback) -> None: """ Thread de fons que escolta passivament els missatges espontanis de la vertebra. No envia cap peticio RTR — no interfereix amb el bus. """ push_id = ARB_ID_DIGITAL_BASE + ARB_ID_DIGITAL_PUSH_OFFSET + i2c_addr last_val_a = None last_val_b = None try: bus = get_bus() while not stop_event.is_set(): msg = bus.recv(timeout=0.2) if msg is None: continue if msg.is_remote_frame: continue if msg.arbitration_id != push_id: continue if len(msg.data) < 5: continue val_a = ~msg.data[3] & 0xFF val_b = ~msg.data[4] & 0xFF # Notificar nomes si hi ha canvi if val_a != last_val_a or val_b != last_val_b: last_val_a = val_a last_val_b = val_b try: callback(i2c_addr, val_a, val_b) except Exception as e: if VERBOSE: print(f"Error en callback on_din_change addr={i2c_addr}: {e}") except Exception as e: if VERBOSE: print(f"Error en listener thread addr={i2c_addr}: {e}") def on_din_change(addr: str, callback) -> None: """ Registra un callback per a canvis d'entrada digital espontanis. El callback s'executa en un thread de fons quan la vertebra publica un canvi. Signatura del callback: callback(addr: int, val_a: int, val_b: int) - addr: adreca i2c de la vertebra (enter, ex: 0) - val_a: byte d'entrades del costat A (0-255) - val_b: byte d'entrades del costat B (0-255) Args: addr: Adreca en format binari (ex: '0000') callback: Funcio a cridar quan hi ha canvi """ i2c_addr = int(addr, 2) off_din_change(addr) # cancel.lar listener anterior si existeix stop_event = threading.Event() t = threading.Thread( target=_din_listener_thread, args=(i2c_addr, stop_event, callback), daemon=True, name=f"din_listener_{i2c_addr}" ) with _din_listeners_lock: _din_listeners[i2c_addr] = (t, stop_event, callback) t.start() if VERBOSE: print(f"on_din_change: listener iniciat per addr={i2c_addr}") def off_din_change(addr: str) -> None: """ Cancel.la el listener de canvis digitals d'una adreca. Args: addr: Adreca en format binari (ex: '0000') """ i2c_addr = int(addr, 2) with _din_listeners_lock: entry = _din_listeners.pop(i2c_addr, None) if entry: t, stop_event, _ = entry stop_event.set() t.join(timeout=1.0) if VERBOSE: print(f"off_din_change: listener aturat per addr={i2c_addr}") def off_all_din_change() -> None: """ Cancel.la tots els listeners de canvis digitals actius. """ with _din_listeners_lock: addrs = list(_din_listeners.keys()) for i2c_addr in addrs: off_din_change(format(i2c_addr, '04b')) # ============================================================================= # AIN ASYNC LISTENER — listener passiu dels valors ADC enviats periodicament # ============================================================================= # La vertebra analogica (firmware >= v1.5, veure set_async_adc()) tramet sola # els 4 valors ADC d'un costat cada period_ms, sense que ningu els demani. # Format observat: # arbitration_id = ARB_ID_ASYNC_PUSH_BASE + i2c_addr + side (side: 0=A, 1=B) # data[i*2] = LSB del canal i (i=0..3) # data[i*2+1] = MSB del canal i # # Diccionari de listeners actius: {(i2c_addr, side_num): (thread, stop_event, callback)} _ain_listeners: dict = {} _ain_listeners_lock = threading.Lock() def _ain_listener_thread(i2c_addr: int, side_num: int, stop_event: threading.Event, callback) -> None: """ Thread de fons que escolta passivament els valors ADC espontanis d'un costat d'una vertebra analogica. No envia cap peticio — no interfereix amb el bus (cal haver activat prevoiament amb set_async_adc()). """ push_id = ARB_ID_ASYNC_PUSH_BASE + i2c_addr + side_num last_values = None try: bus = get_bus() while not stop_event.is_set(): msg = bus.recv(timeout=0.2) if msg is None: continue if msg.is_remote_frame: continue if msg.arbitration_id != push_id: continue if len(msg.data) < 8: continue values = [msg.data[i * 2] | (msg.data[i * 2 + 1] << 8) for i in range(4)] if values != last_values: last_values = values try: callback(i2c_addr, 'A' if side_num == 0 else 'B', values) except Exception as e: if VERBOSE: print(f"Error en callback on_ain_change addr={i2c_addr}: {e}") except Exception as e: if VERBOSE: print(f"Error en listener thread ain addr={i2c_addr}: {e}") def on_ain_change(addr: str, side: str, callback) -> None: """ Registra un callback per als valors ADC enviats periodicament per una vertebra analogica. Cal haver activat prevoiament l'enviament amb set_async_adc(addr, side, True, period_ms). Signatura del callback: callback(addr: int, side: str, values: list[int]) - addr: adreca i2c de la vertebra (enter, ex: 0) - side: 'A' o 'B' - values: llista de 4 valors ADC (0-26624). Usa ain2v() per convertir a volts. Args: addr: Adreca en format binari (ex: '0000') side: 'A' o 'B' (un listener per costat) callback: Funcio a cridar quan arriben nous valors """ i2c_addr = int(addr, 2) side_num = 0 if side.lower() == 'a' else 1 off_ain_change(addr, side) # cancel.lar listener anterior si existeix stop_event = threading.Event() t = threading.Thread( target=_ain_listener_thread, args=(i2c_addr, side_num, stop_event, callback), daemon=True, name=f"ain_listener_{i2c_addr}_{side}" ) with _ain_listeners_lock: _ain_listeners[(i2c_addr, side_num)] = (t, stop_event, callback) t.start() if VERBOSE: print(f"on_ain_change: listener iniciat per addr={i2c_addr}, side={side}") def off_ain_change(addr: str, side: str) -> None: """ Cancel.la el listener de valors ADC asincrons d'un costat d'una vertebra. Args: addr: Adreca en format binari (ex: '0000') side: 'A' o 'B' """ i2c_addr = int(addr, 2) side_num = 0 if side.lower() == 'a' else 1 with _ain_listeners_lock: entry = _ain_listeners.pop((i2c_addr, side_num), None) if entry: t, stop_event, _ = entry stop_event.set() t.join(timeout=1.0) if VERBOSE: print(f"off_ain_change: listener aturat per addr={i2c_addr}, side={side}") def off_all_ain_change() -> None: """ Cancel.la tots els listeners de valors ADC asincrons actius. """ with _ain_listeners_lock: keys = list(_ain_listeners.keys()) for i2c_addr, side_num in keys: off_ain_change(format(i2c_addr, '04b'), 'A' if side_num == 0 else 'B') # ============================================================================= # GPIO SETUP (LED INDICATORS) # ============================================================================= if USE_LEDS: GPIO.setmode(GPIO.BCM) GPIO.setwarnings(False) GPIO.setup(LED_W, GPIO.OUT) GPIO.setup(LED_B, GPIO.OUT) # ============================================================================= # CONFIGURATION DECODERS # ============================================================================= def dvert_cfg_from_number(n: int) -> str: """ Converteix el número de configuració de vertebra digital a string descriptiu. Args: n: Número de configuració Returns: String amb la configuració (ex: "A:din, B:dout") """ config_map = { 17: "A:din, B:din", 33: "A:din, B:dout", 18: "A:dout, B:din", 34: "A:dout, B:dout", 20: "A:pwm, B:din", 36: "A:pwm, B:dout", 65: "A:din, B:pwm", 66: "A:dout, B:pwm", 130: "A:dout, B:touch", 129: "A:din, B:touch", 132: "A:pwm, B:touch" } return config_map.get(n, "A:?, B:?") def avert_cfg_from_number(by_rib_a: int, by_rib_b: int) -> str: """ Converteix els bytes de configuració de vertebra analògica a string descriptiu. Args: by_rib_a: Byte de configuració del rib A by_rib_b: Byte de configuració del rib B Returns: String amb la configuració (ex: "A:ain, B:aout") """ def get_mode(by_rib: int) -> str: if by_rib == 1: return "ain" elif by_rib == 2: return "aout" else: return "?" return f"A:{get_mode(by_rib_a)}, B:{get_mode(by_rib_b)}" # ============================================================================= # VOLTAGE CONVERSION UTILITIES # ============================================================================= def ain2v(ain_value: int) -> float: """ Converteix valor analògic d'entrada (0-26624) a voltatge (-10V a +10V). Args: ain_value: Valor digital llegit (0-26624) Returns: Voltatge en volts amb 2 decimals """ return round((((20 * float(ain_value)) / AIN_MAX_VALUE) + VOLTAGE_MIN) * 100) / 100 def v2aout(voltage_0_10: float) -> int: """ Converteix voltatge (0-10V) a valor digital per sortida analògica. Args: voltage_0_10: Voltatge entre 0 i 10V Returns: Valor digital entre 0 i 4095 """ ret_v = round((voltage_0_10 * AOUT_MAX_VALUE) / AOUT_VOLTAGE_MAX) return max(0, min(AOUT_MAX_VALUE, ret_v)) # ============================================================================= # CAN MESSAGE BUILDERS # ============================================================================= def ain_msg(addr: str, side: str) -> can.Message: """ Crea missatge CAN per llegir entrada analògica. Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' Returns: Missatge CAN configurat """ i2c_addr = int(addr, 2) offset = 0 if side.lower() == 'a' else 256 return can.Message( arbitration_id=ARB_ID_ANALOG_BASE + i2c_addr + offset, data=[], is_extended_id=False, is_remote_frame=True, dlc=0, timestamp=datetime.timestamp(datetime.now()) ) def aout_msg(addr: str, side: str, ndac: int, value: int) -> can.Message: """ Crea missatge CAN per escriure sortida analògica. Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' ndac: Número de DAC (1, 2, 3 o 4) value: Valor entre 0 i 4095 (0V a 10V) Returns: Missatge CAN configurat """ i2c_addr = int(addr, 2) # Calcular el byte de comandament segons el side i ndac base_cmd = 2 if side.lower() == 'a' else 18 ndac_offset = {1: 0, 2: 32, 3: 64, 4: 96}.get(ndac, 0) msg_data = [ base_cmd + ndac_offset, value // 256, value % 256 ] if VERBOSE: print(f"AOUT message: {msg_data}") return can.Message( arbitration_id=ARB_ID_AOUT_BASE + i2c_addr, data=msg_data, is_extended_id=False, is_remote_frame=False, dlc=3, timestamp=datetime.timestamp(datetime.now()) ) def async_adc_msg(addr: str, side: str, enable: bool, period_ms: int = 50) -> can.Message: """ Crea missatge CAN per activar/desactivar la transmissio periodica (asincrona) dels valors ADC d'una vertebra analogica. Args: addr: Adreca en format binari (ex: '0000') side: 'A', 'B' o 'AB' (ambdos costats). Ignorat si enable=False i es vol desactivar tot (usa side='AB'). enable: True per activar, False per desactivar period_ms: Periode d'enviament en ms (el firmware el limita a 10-5000ms) Returns: Missatge CAN configurat Exemple: # Enviar els 4 canals ADC del costat A cada 200ms send_can(bus, async_adc_msg('0000', 'A', True, 200)) # Desactivar l'enviament asincron dels dos costats send_can(bus, async_adc_msg('0000', 'AB', False)) """ i2c_addr = int(addr, 2) side_bits = {'a': 0x01, 'b': 0x02, 'ab': 0x03, 'ba': 0x03}.get(side.lower(), 0x03) period_ms = max(10, min(5000, period_ms)) msg_data = [ ASYNC_CFG, side_bits, 1 if enable else 0, period_ms & 0xFF, (period_ms >> 8) & 0xFF ] if VERBOSE: print(f"ASYNC_CFG message: {msg_data}") return can.Message( arbitration_id=ARB_ID_ASYNC_CFG_BASE + i2c_addr, data=msg_data, is_extended_id=False, is_remote_frame=False, dlc=5, timestamp=datetime.timestamp(datetime.now()) ) def dout_msg(addr: str, side: str, value: int) -> can.Message: """ Crea missatge CAN per escriure sortida digital (byte complet). Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' value: Valor del byte (0-255) Returns: Missatge CAN configurat """ i2c_addr = int(addr, 2) msg_data = [2 if side.lower() == 'a' else 18, value] if VERBOSE: print(f"DOUT message: {msg_data}") return can.Message( arbitration_id=ARB_ID_DIGITAL_BASE + i2c_addr, data=msg_data, is_extended_id=False, is_remote_frame=False, dlc=2, timestamp=datetime.timestamp(datetime.now()) ) def doutbit_msg(addr: str, side: str, posbyte: int, value: int) -> can.Message: """ Crea missatge CAN per escriure un bit específic de sortida digital. Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' posbyte: Posició del bit (0-7) value: Valor del bit (0 o 1) Returns: Missatge CAN configurat """ i2c_addr = int(addr, 2) base_val = 10 if side.lower() == 'a' else 26 val = base_val + 32 * posbyte msg_data = [val, value] if VERBOSE: print(f"DOUTBIT message: {msg_data}") return can.Message( arbitration_id=ARB_ID_DIGITAL_BASE + i2c_addr, data=msg_data, is_extended_id=False, is_remote_frame=False, dlc=2, timestamp=datetime.timestamp(datetime.now()) ) def doutpwm_msg(addr: str, side: str, bit: int, value: int) -> can.Message: """ Crea missatge CAN per escriure PWM (0-255) a un bit concret de sortida digital. La vertebra ha d'estar configurada com a 'pwm' a aquell costat (dsetup) — només un costat (A o B) pot ser PWM alhora. Mapa de registres confirmat amb tres fonts independents (i2c_iotv.py, IoTVertebrae_Arduino.zip::_doutpwmMsg(), i el propi firmware dvert_v3_asyncInputChanges_03.ino — bits del byReg: bit3=escriptura de bit (1) vs byte (0), bit4=costat (0=A,1=B), bits5-7=número de bit/canal): REG_PWM (0x03) | 0x08 (nivell de bit) | (0x10 si B) + 0x20*bit = 0x0B + 32*bit (costat A) / 0x1B + 32*bit (costat B). Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' bit: Posició del bit/canal (0-7) value: Cicle de treball PWM (0-255) Returns: Missatge CAN configurat """ i2c_addr = int(addr, 2) base_val = 11 if side.lower() == 'a' else 27 val = base_val + 32 * bit msg_data = [val, value] if VERBOSE: print(f"DOUTPWM message: {msg_data}") return can.Message( arbitration_id=ARB_ID_DIGITAL_BASE + i2c_addr, data=msg_data, is_extended_id=False, is_remote_frame=False, dlc=2, timestamp=datetime.timestamp(datetime.now()) ) def din_msg(addr: str) -> can.Message: """ Crea missatge CAN per llegir entrada digital. Args: addr: Adreça en format binari (ex: '0000') Returns: Missatge CAN configurat """ i2c_addr = int(addr, 2) return can.Message( arbitration_id=ARB_ID_DIGITAL_BASE + i2c_addr, data=[], is_extended_id=False, is_remote_frame=True, dlc=0, timestamp=datetime.timestamp(datetime.now()) ) def dversion_msg(addr: str) -> can.Message: """ Crea missatge CAN per llegir versió de vertebra digital. Args: addr: Adreça en format binari (ex: '0000') Returns: Missatge CAN configurat """ i2c_addr = int(addr, 2) return can.Message( arbitration_id=ARB_ID_DIGITAL_BASE + i2c_addr, data=[], is_extended_id=False, is_remote_frame=True, dlc=0, timestamp=datetime.timestamp(datetime.now()) ) def aversion_msg(addr: str) -> can.Message: """ Crea missatge CAN per llegir versió de vertebra analògica. Args: addr: Adreça en format binari (ex: '0000') Returns: Missatge CAN configurat """ i2c_addr = int(addr, 2) return can.Message( arbitration_id=ARB_ID_ANALOG_BASE + i2c_addr + ARB_ID_ANALOG_VERSION, data=[], is_extended_id=False, is_remote_frame=True, dlc=0, timestamp=datetime.timestamp(datetime.now()) ) # ============================================================================= # LOW-LEVEL CAN SEND/RECEIVE # ============================================================================= def send_can(bus: can.interface.Bus, msg: can.Message) -> None: """ Envia un missatge pel bus CAN amb indicador LED opcional. Args: bus: Instància del bus CAN msg: Missatge a enviar """ if USE_LEDS: GPIO.output(LED_W, GPIO.HIGH) try: bus.send(msg) sleep(0.001) # Evitar llegir el que s'acaba de trametre except can.CanError as e: if VERBOSE: print(f"Error sending CAN message: {e}") raise finally: if USE_LEDS: GPIO.output(LED_W, GPIO.LOW) def recv_can(bus: can.interface.Bus) -> Optional[can.Message]: """ Rep un missatge del bus CAN amb timeout i indicador LED opcional. Args: bus: Instància del bus CAN Returns: Missatge rebut o None si timeout """ if USE_LEDS: GPIO.output(LED_B, GPIO.HIGH) try: msg = bus.recv(TIMEOUT_RCV) return msg except can.CanError as e: if VERBOSE: print(f"Error receiving CAN message: {e}") return None finally: if USE_LEDS: GPIO.output(LED_B, GPIO.LOW) # ============================================================================= # HIGH-LEVEL API FUNCTIONS # ============================================================================= def ain(addr: str, side: str, ndac: int) -> Union[int, str]: """ Llegeix entrada analògica. Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' ndac: Número de DAC a llegir (1, 2, 3 o 4) Returns: Valor digital llegit (0-26624) o "Error" si falla """ try: bus = get_bus() send_can(bus, ain_msg(addr, side)) msg = recv_can(bus) if VERBOSE and msg: print(f"AIN response: {msg}") print(f" is_remote_frame: {msg.is_remote_frame}") print(f" data: {msg.data}") print(f" dlc: {msg.dlc}") if msg is None: if VERBOSE: print(f"No CAN answer in {TIMEOUT_RCV} seconds") return "Error" pos = ndac * 2 - 2 if pos + 1 >= len(msg.data): if VERBOSE: print(f"Insufficient data in message for DAC {ndac}") return "Error" return 256 * msg.data[pos] + msg.data[pos + 1] except Exception as e: if VERBOSE: print(f"Error in ain(): {e}") return "Error" def ainBatch(addr: str, side: str) -> Union[list, str]: """ Llegeix els 4 canals analògics d'entrada EN UNA SOLA petició CAN. ain() ja fa una petició RTR que la vertebra respon amb els 4 canals sencers en un unic missatge (nomes en descarta 3 i en retorna 1); si cal mes d'un canal, cridar ain() un cop per canal fa un round-trip CAN complet per cadascun, redemanant i descartant els altres 3 cada vegada. ainBatch() fa la mateixa petició UNA sola vegada i retorna els 4 valors. Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' Returns: Llista de 4 valors (0-26624), un per canal (index 0 = canal 1, index 1 = canal 2, etc.), o "Error" si falla. """ try: bus = get_bus() send_can(bus, ain_msg(addr, side)) msg = recv_can(bus) if VERBOSE and msg: print(f"AIN batch response: {msg}") print(f" is_remote_frame: {msg.is_remote_frame}") print(f" data: {msg.data}") print(f" dlc: {msg.dlc}") if msg is None: if VERBOSE: print(f"No CAN answer in {TIMEOUT_RCV} seconds") return "Error" if len(msg.data) < 8: if VERBOSE: print("Insufficient data in message for ainBatch") return "Error" return [256 * msg.data[ch * 2] + msg.data[ch * 2 + 1] for ch in range(4)] except Exception as e: if VERBOSE: print(f"Error in ainBatch(): {e}") return "Error" def din(addr: str, side: str) -> Union[str, str]: """ Llegeix entrada digital (8 bits). Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' Returns: String de 8 bits (ex: '10101010') o "Error" si falla """ try: bus = get_bus() send_can(bus, din_msg(addr)) msg = recv_can(bus) if VERBOSE and msg: print(f"DIN response: {msg}") print(f" is_remote_frame: {msg.is_remote_frame}") print(f" data: {msg.data}") print(f" dlc: {msg.dlc}") if msg is None: if VERBOSE: print(f"No CAN answer in {TIMEOUT_RCV} seconds") return "Error" pos = 3 if side.lower() == 'a' else 4 if pos >= len(msg.data): if VERBOSE: print(f"Insufficient data in message") return "Error" valor = msg.data[pos] # Invertir els bits i formatejar com a string binari de 8 bits return format(~valor & 255, 'b').zfill(8) except Exception as e: if VERBOSE: print(f"Error in din(): {e}") return "Error" def aout(addr: str, side: str, ndac: int, value: int) -> None: """ Escriu sortida analògica. Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' ndac: Número de DAC (1, 2, 3 o 4) value: Valor entre 0 i 4095 (0V a 10V) """ try: bus = get_bus() send_can(bus, aout_msg(addr, side, ndac, value)) except Exception as e: if VERBOSE: print(f"Error in aout(): {e}") def set_async_adc(addr: str, side: str, enable: bool, period_ms: int = 50) -> None: """ Activa o desactiva l'enviament periodic (asincron) dels valors ADC d'una vertebra analogica, i en configura el periode. Aixo evita haver de fer polling (ain()) des del cap: la vertebra envia sola els 4 valors ADC del costat indicat cada period_ms mil.lisegons. Cal firmware >= v1.5 (registre ASYNC_CFG). Args: addr: Adreca en format binari (ex: '0000') side: 'A', 'B' o 'AB' enable: True per activar, False per desactivar period_ms: Periode en ms (limitat pel firmware entre 10 i 5000ms) Exemple: set_async_adc('0000', 'A', True, 200) # cada 200ms, costat A set_async_adc('0000', 'AB', False) # desactiva els dos costats """ try: bus = get_bus() send_can(bus, async_adc_msg(addr, side, enable, period_ms)) if VERBOSE: estat = "activat" if enable else "desactivat" print(f"Async ADC {estat} per addr={addr}, side={side}, period={period_ms}ms") except Exception as e: if VERBOSE: print(f"Error in set_async_adc(): {e}") def dout(addr: str, side: str, value: int) -> None: """ Escriu sortida digital (byte complet). Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' value: Valor del byte (0-255) """ try: bus = get_bus() send_can(bus, dout_msg(addr, side, value)) except Exception as e: if VERBOSE: print(f"Error in dout(): {e}") def doutbit(addr: str, side: str, posbyte: int, value: int) -> None: """ Escriu un bit específic de sortida digital. Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' posbyte: Posició del bit (0-7) value: Valor del bit (0 o 1) """ try: bus = get_bus() send_can(bus, doutbit_msg(addr, side, posbyte, value)) except Exception as e: if VERBOSE: print(f"Error in doutbit(): {e}") def doutpwm(addr: str, side: str, bit: int, value: int) -> None: """ Escriu un cicle de treball PWM (0-255) a un bit/canal concret. Cal que aquell costat (A o B) estigui configurat com a 'pwm' amb dsetup() primer — només un dels dos costats pot ser PWM alhora. Args: addr: Adreça en format binari (ex: '0000') side: 'A' o 'B' bit: Posició del bit/canal (0-7) value: Cicle de treball PWM (0-255) """ try: bus = get_bus() send_can(bus, doutpwm_msg(addr, side, bit, value)) except Exception as e: if VERBOSE: print(f"Error in doutpwm(): {e}") def dsetup(addr: str, mode_a: str, mode_b: str) -> None: """ Configura els modes de treball d'una vertebra digital. Restriccions: - PWM: Només un rib (A o B) pot gestionar PWM - Touch: Només el rib B pot gestionar 8 entrades tàctils Args: addr: Adreça en format binari (ex: '0000') mode_a: Mode del rib A ('din', 'dout', 'pwm') mode_b: Mode del rib B ('din', 'dout', 'pwm', 'touch') """ try: bus = get_bus() i2c_addr = int(addr, 2) # Validació: no es pot tenir PWM als dos costats if mode_a.lower() == 'pwm' and mode_b.lower() == 'pwm': print("Error: PWM només pot estar a un costat (A o B)") return # Validació: touch només pot estar a B if mode_a.lower() == 'touch': print("Error: Touch només s'accepta al costat B") return # Configurar direccions dels ports d1 = 0 if mode_a.lower() in ['dout', 'pwm'] else 255 d2 = 0 if mode_b.lower() in ['dout', 'pwm'] else 255 msg_data = [0, d1, d2] if VERBOSE: print(f"Setup message: {msg_data}") can_msg = can.Message( arbitration_id=ARB_ID_DIGITAL_BASE + i2c_addr, data=msg_data, is_extended_id=False, is_remote_frame=False, dlc=3, timestamp=datetime.timestamp(datetime.now()) ) send_can(bus, can_msg) # Configurar PWM si cal if mode_a.lower() == 'pwm': msg_data = [4, 0] can_msg = can.Message( arbitration_id=ARB_ID_DIGITAL_BASE + i2c_addr, data=msg_data, is_extended_id=False, is_remote_frame=False, dlc=2, timestamp=datetime.timestamp(datetime.now()) ) send_can(bus, can_msg) elif mode_b.lower() == 'pwm': msg_data = [4, 1] can_msg = can.Message( arbitration_id=ARB_ID_DIGITAL_BASE + i2c_addr, data=msg_data, is_extended_id=False, is_remote_frame=False, dlc=2, timestamp=datetime.timestamp(datetime.now()) ) send_can(bus, can_msg) # Configurar Touch si cal if mode_b.lower() == 'touch': msg_data = [7, 1] can_msg = can.Message( arbitration_id=ARB_ID_DIGITAL_BASE + i2c_addr, data=msg_data, is_extended_id=False, is_remote_frame=False, dlc=2, timestamp=datetime.timestamp(datetime.now()) ) send_can(bus, can_msg) if VERBOSE: print(f"Configuration set: A={mode_a}, B={mode_b}") except Exception as e: if VERBOSE: print(f"Error in dsetup(): {e}") def dversion(addr: str) -> str: """ Llegeix la versió del firmware d'una vertebra digital. Args: addr: Adreça en format binari (ex: '0000') Returns: String amb la versió (ex: '1.5') o '0.0' si error """ try: bus = get_bus() send_can(bus, dversion_msg(addr)) msg = recv_can(bus) if VERBOSE and msg: print(f"DVERSION response: {msg}") if msg is not None and len(msg.data) >= 3: version = f"{msg.data[1]}.{msg.data[2]}" if VERBOSE: cfg = dvert_cfg_from_number(msg.data[0]) print(f"Digital vert {addr} --> Cfg: {cfg}. Version: {version}") return version else: if VERBOSE: print(f"No CAN answer in {TIMEOUT_RCV} seconds") return "0.0" except Exception as e: if VERBOSE: print(f"Error in dversion(): {e}") return "0.0" def getdsetup(addr: str) -> str: """ Llegeix la configuració actual d'una vertebra digital. Args: addr: Adreça en format binari (ex: '0000') Returns: String amb la configuració (ex: 'A:din, B:dout') o 'A:?, B:?' si error """ try: bus = get_bus() send_can(bus, dversion_msg(addr)) msg = recv_can(bus) if VERBOSE and msg: print(f"GETDSETUP response: {msg}") if msg is not None and len(msg.data) >= 3: dsetup = dvert_cfg_from_number(msg.data[0]) version = f"{msg.data[1]}.{msg.data[2]}" if VERBOSE: print(f"Digital vert {addr} --> Cfg: {dsetup}. Version: {version}") return dsetup else: if VERBOSE: print(f"No CAN answer in {TIMEOUT_RCV} seconds") return "A:?, B:?" except Exception as e: if VERBOSE: print(f"Error in getdsetup(): {e}") return "A:?, B:?" def aversion(addr: str) -> str: """ Llegeix la versió del firmware d'una vertebra analògica. Args: addr: Adreça en format binari (ex: '0000') Returns: String amb la versió (ex: '1.4') o '0.0' si error """ try: bus = get_bus() send_can(bus, aversion_msg(addr)) msg = recv_can(bus) if VERBOSE and msg: print(f"AVERSION response: {msg}") if msg is not None and len(msg.data) >= 4: version = f"{msg.data[1]}.{msg.data[2]}" if VERBOSE: cfg = avert_cfg_from_number(msg.data[0], msg.data[3]) print(f"Analog vert {addr} --> Cfg: {cfg}. Version: {version}") return version else: if VERBOSE: print(f"No CAN answer in {TIMEOUT_RCV} seconds") return "0.0" except Exception as e: if VERBOSE: print(f"Error in aversion(): {e}") return "0.0" def getasetup(addr: str) -> str: """ Llegeix la configuració actual d'una vertebra analògica. Args: addr: Adreça en format binari (ex: '0000') Returns: String amb la configuració (ex: 'A:ain, B:aout') o 'A:?, B:?' si error """ try: bus = get_bus() send_can(bus, aversion_msg(addr)) msg = recv_can(bus) if VERBOSE and msg: print(f"GETASETUP response: {msg}") if msg is not None and len(msg.data) >= 4: asetup = avert_cfg_from_number(msg.data[0], msg.data[3]) version = f"{msg.data[1]}.{msg.data[2]}" if VERBOSE: print(f"Analog vert {addr} --> Cfg: {asetup}. Version: {version}") return asetup else: if VERBOSE: print(f"No CAN answer in {TIMEOUT_RCV} seconds") return "A:?, B:?" except Exception as e: if VERBOSE: print(f"Error in getasetup(): {e}") return "A:?, B:?" # ============================================================================= # CAN INTERFACE CONTROL # ============================================================================= def can_on() -> None: """ Activa la interfície CAN a la Raspberry Pi. Configura el bitrate a 100kbps. """ try: os.system(f'sudo /sbin/ip link set up {CAN_CHANNEL} type can bitrate {CAN_BITRATE}') sleep(0.1) if VERBOSE: print(f"CAN interface {CAN_CHANNEL} activated at {CAN_BITRATE} bps") except Exception as e: print(f"Error activating CAN interface: {e}") def can_off() -> None: """ Desactiva la interfície CAN. """ try: # Tancar el bus abans de desactivar la interfície close_bus() os.system(f'sudo /sbin/ifconfig {CAN_CHANNEL} down') if VERBOSE: print(f"CAN interface {CAN_CHANNEL} deactivated") except Exception as e: print(f"Error deactivating CAN interface: {e}") # ============================================================================= # MAIN - TEST CODE # ============================================================================= if __name__ == "__main__": try: espera = 0.5 can_on() # Test: Activar bits individualment en seqüència print("Testing bit-by-bit output...") for i in range(10): for bit in range(8): doutbit('0000', 'A', bit, 1) sleep(0.05) doutbit('0000', 'A', bit, 0) print("\nTest completed successfully!") except KeyboardInterrupt: print("\nInterrupted by user") dout('0000', 'A', 0x00) print("Digital output cleared") print(f"Digital input B: {din('0000', 'B')}") except Exception as e: print(f"Error during execution: {e}") finally: # Assegurar el tancament correcte close_bus() can_off() print("CAN bus closed and interface deactivated")