Message Handling

class Message:
    def __init__(self, timestamp=0.0, arbitration_id=0, is_extended_id=True, 
                 is_remote_frame=False, is_error_frame=False, channel=None,
                 dlc=None, data=None, is_fd=False, is_rx=True, 
                 bitrate_switch=False, error_state_indicator=False, check=False):
        """
        Create a CAN message object.
        
        Parameters:
        - timestamp: Message timestamp in seconds (float)
        - arbitration_id: CAN arbitration ID (0-0x1FFFFFFF for extended, 0-0x7FF for standard)
        - is_extended_id: True for 29-bit extended ID, False for 11-bit standard ID
        - is_remote_frame: True for remote transmission request frames
        - is_error_frame: True for error frames
        - channel: Channel identifier (backend specific)
        - dlc: Data Length Code (0-8 for CAN 2.0, 0-64 for CAN FD)
        - data: Message data as bytes, bytearray, or list of integers
        - is_fd: True for CAN FD frames
        - is_rx: True for received messages, False for transmitted
        - bitrate_switch: True for CAN FD bitrate switching
        - error_state_indicator: True for CAN FD error state indicator
        - check: If True, validate message parameters (raises ValueError on invalid)
        
        Raises:
        - ValueError: Invalid message parameters (only if check=True)
        - TypeError: Invalid data type
        """

# Message properties (all read/write)
timestamp: float                # Message timestamp in seconds
arbitration_id: int            # CAN arbitration ID
is_extended_id: bool           # Extended (29-bit) vs standard (11-bit) ID
is_remote_frame: bool          # Remote transmission request
is_error_frame: bool           # Error frame indicator  
channel: Any                   # Channel identifier
dlc: int                       # Data Length Code
data: bytearray               # Message data bytes
is_fd: bool                   # CAN FD frame
is_rx: bool                   # Receive (True) vs transmit (False)
bitrate_switch: bool          # CAN FD bitrate switching
error_state_indicator: bool   # CAN FD error state indicator

def equals(self, other: Message, timestamp_delta=1e-6, check_channel=True, 
          check_direction=True) -> bool:
    """
    Compare this message with another message.
    
    Parameters:
    - other: Message to compare with
    - timestamp_delta: Maximum timestamp difference in seconds (None to ignore)
    - check_channel: Whether to compare message channels
    - check_direction: Whether to compare message directions (Rx/Tx)
    
    Returns:
    True if messages are equal within specified tolerances
    """

def _check(self) -> None:
    """
    Check if message parameters are valid.
    
    Validates:
    - Timestamp is not negative, infinite, or NaN
    - Remote and error frames are mutually exclusive
    - CAN FD compatibility with remote frames
    - Arbitration ID ranges (11-bit: 0-0x7FF, 29-bit: 0-0x1FFFFFFF)
    - DLC ranges (CAN 2.0: 0-8, CAN FD: 0-64)
    - Data consistency with DLC and frame type
    - CAN FD specific flags only used with CAN FD frames
    
    Raises:
    - ValueError: Invalid message parameters
    """

def __str__(self) -> str:
    """
    Return human-readable string representation.
    
    Format: Timestamp ID Flags DL Data ASCII Channel
    Example: "    12.345678    ID: 123    S Rx     DL:  8    01 02 03 04 05 06 07 08    'test'"
    """

def __repr__(self) -> str:
    """
    Return unambiguous string representation suitable for recreating the message.
    
    Example: "can.Message(timestamp=12.345, arbitration_id=0x123, ...)"
    """

def __format__(self, format_spec: str) -> str:
    """Format message using format specification (empty spec uses __str__)."""

def __len__(self) -> int:
    """Return the DLC (works for remote frames)."""

def __bytes__(self) -> bytes:
    """Return message data as bytes."""

def __bool__(self) -> bool:
    """Messages are always truthy."""

def __copy__(self) -> Message:
    """Create a shallow copy of the message."""

def __deepcopy__(self, memo) -> Message:
    """Create a deep copy of the message with all referenced objects copied."""

import can

# Standard 11-bit ID message
msg1 = can.Message(
    arbitration_id=0x123,
    data=[0x11, 0x22, 0x33, 0x44],
    is_extended_id=False
)

# Extended 29-bit ID message
msg2 = can.Message(
    arbitration_id=0x18FF1234,
    data=b'\x01\x02\x03\x04\x05\x06\x07\x08',
    is_extended_id=True
)

# Remote frame
remote_msg = can.Message(
    arbitration_id=0x456,
    is_remote_frame=True,
    dlc=8,  # Request 8 bytes
    is_extended_id=False
)

print(f"Standard message: {msg1}")
print(f"Extended message: {msg2}")
print(f"Remote frame: {remote_msg}")

import can

# CAN FD message with up to 64 bytes
fd_msg = can.Message(
    arbitration_id=0x123,
    data=list(range(64)),  # 64 bytes of data
    is_fd=True,
    bitrate_switch=True,
    is_extended_id=False
)

print(f"CAN FD message length: {len(fd_msg)}")
print(f"CAN FD data: {fd_msg.data[:10]}...")  # First 10 bytes

import can

try:
    # This will raise ValueError because remote frames can't carry data
    invalid_msg = can.Message(
        arbitration_id=0x123,
        data=[1, 2, 3, 4],
        is_remote_frame=True,
        check=True  # Enable validation
    )
except ValueError as e:
    print(f"Validation error: {e}")

# Valid message creation
valid_msg = can.Message(
    arbitration_id=0x123,
    data=[1, 2, 3, 4],
    check=True
)

import can
import time

# Create two similar messages
msg1 = can.Message(
    timestamp=time.time(),
    arbitration_id=0x123,
    data=[1, 2, 3, 4]
)

time.sleep(0.001)  # Small delay

msg2 = can.Message(
    timestamp=time.time(),
    arbitration_id=0x123,
    data=[1, 2, 3, 4]
)

# Compare with default timestamp tolerance (1µs)
if msg1.equals(msg2):
    print("Messages are equal (within timestamp tolerance)")

# Compare ignoring timestamps
if msg1.equals(msg2, timestamp_delta=None):
    print("Messages have identical content")

# Strict comparison (identity)
if msg1 is msg2:
    print("Same message object")
else:
    print("Different message objects")

import can

msg = can.Message(
    arbitration_id=0x123,
    data=[0x48, 0x65, 0x6C, 0x6C, 0x6F]  # "Hello" in ASCII
)

# Access data in different ways
print(f"Data as list: {list(msg.data)}")
print(f"Data as bytes: {bytes(msg)}")
print(f"Data as hex: {msg.data.hex()}")
print(f"Data length: {len(msg)}")

# Modify data
msg.data[0] = 0x48  # Set first byte
msg.data.extend([0x21])  # Add exclamation mark
msg.dlc = len(msg.data)  # Update DLC

print(f"Modified message: {msg}")

import can
import copy

original = can.Message(
    arbitration_id=0x123,
    data=[1, 2, 3, 4],
    channel="can0"
)

# Shallow copy
shallow_copy = copy.copy(original)
shallow_copy.arbitration_id = 0x456

# Deep copy
deep_copy = copy.deepcopy(original)
deep_copy.data[0] = 99

print(f"Original: {original}")
print(f"Shallow copy: {shallow_copy}")
print(f"Deep copy: {deep_copy}")

from typing import Union, Optional, Any
from collections.abc import Sequence

# Type aliases for message data
CanData = Union[bytes, bytearray, Sequence[int]]
Channel = Union[str, int, Any]

class Message:
    """CAN message representation with support for all CAN variants."""
    
    # Slots for memory efficiency
    __slots__ = (
        "__weakref__",
        "arbitration_id", "bitrate_switch", "channel", "data", "dlc",
        "error_state_indicator", "is_error_frame", "is_extended_id", 
        "is_fd", "is_remote_frame", "is_rx", "timestamp"
    )