# nxt.brick module -- Classes to represent LEGO Mindstorms NXT bricks # Copyright (C) 2006 Douglas P Lau # Copyright (C) 2009 Marcus Wanner, rhn # Copyright (C) 2010 rhn, Marcus Wanner, zonedabone # Copyright (C) 2021 Nicolas Schodet # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. import io import sys import threading import time from collections.abc import Iterator from types import TracebackType from typing import IO, Any, Optional, cast import nxt.error import nxt.motor import nxt.sensor import nxt.sensor.digital from nxt.telegram import Opcode, Telegram __all__ = ["Brick"] # No Buffer before 3.12. if sys.version_info >= (3, 12): from collections.abc import Buffer else: from typing import Any as Buffer class RawFileReader(io.RawIOBase): """Implement RawIOBase for reading a file on the NXT brick.""" def __init__(self, brick: "Brick", name: str) -> None: self._brick = brick self._handle, self._remaining = brick.file_open_read(name) def close(self) -> None: if not self.closed: super().close() self._brick.file_close(self._handle) def readable(self) -> bool: return True def readinto(self, b: Buffer) -> int: rsize = min(self._brick._sock.bsize, self._remaining, len(b)) if rsize == 0: return 0 _, data = self._brick.file_read(self._handle, rsize) size = len(data) self._remaining -= size b[0:size] = data return size class RawFileWriter(io.RawIOBase): """Implement RawIOBase for writing a file on the NXT brick.""" def __init__(self, brick: "Brick", name: str, size: int) -> None: self._brick = brick self._handle = brick.file_open_write(name, size) self._remaining = size def close(self) -> None: if not self.closed: super().close() self._brick.file_close(self._handle) def writable(self) -> bool: return True def write(self, b: Buffer) -> int: if self.closed: raise ValueError("write to closed file") if self._remaining == 0: raise ValueError("write to a full file") wsize = min(self._brick._sock.bsize, self._remaining, len(b)) _, size = self._brick.file_write(self._handle, bytes(b[:wsize])) self._remaining -= size return size class Brick: """Object connected to a NXT brick. It provides low level access to brick commands and high level access to internal brick components (file system, modules...). It can be used to create motor or sensor objects. Create an instance with :func:`nxt.locator.find`. The :class:`Brick` object implements the context manager interface, so you can use it with the ``with`` syntax to close the connection when done with it. """ def __init__(self, sock) -> None: self._sock = sock self._lock = threading.Lock() def play_tone_and_wait(self, frequency_hz: int, duration_ms: int) -> None: """Play a tone and wait until finished. :param frequency_hz: Tone frequency in Hertz. :param duration_ms: Tone duration in milliseconds. """ self.play_tone(frequency_hz, duration_ms) time.sleep(duration_ms / 1000.0) def close(self) -> None: """Disconnect from the NXT brick.""" if self._sock is not None: self._sock.close() self._sock = None def __enter__(self) -> "Brick": return self def __exit__( self, exc_type: Optional[type[BaseException]], exc_value: Optional[BaseException], traceback: Optional[TracebackType], ) -> None: self.close() def __del__(self) -> None: self.close() def open_file( self, name: str, mode: str = "r", size: Optional[int] = None, *, buffering: int = -1, encoding: Optional[str] = None, errors: Optional[str] = None, newline: Optional[str] = None, ) -> io.IOBase: """Open a file and return a corresponding file-like object. :param name: Name of the file to open. :param mode: Specification of open mode. :param size: For writing, give the final size of the file. :param buffering: Buffering control. :param encoding: Encoding for text mode. :param errors: Encoding error handling for text mode. :param newline: Newline handling for text mode. :return: A file-like object connected to the file on the NXT brick. :raises nxt.error.FileNotFoundError: When file does not exists. :raises nxt.error.FileExistsError: When file already exists. :raises nxt.error.SystemProtocolError: When no space is available. `mode` is a string which specifies how the file should be open. You can combine several characters to build the specification: ========= ===================================== Character Meaning ========= ===================================== 'r' open for reading (default) 'w' open for writing (`size` must be given) 't' use text mode (default) 'b' use binary mode ========= ===================================== When writing a file, the NXT brick needs to know the total size when opening the file, so this must be given as parameter. Other parameters (`buffering`, `encoding`, `errors` and `newline`) have the same meaning as the standard :func:`open` function, they must be given as keyword parameters. When `encoding` is ``None`` or not given, it defaults to ``ascii`` as this is the only encoding understood by the NXT brick. """ rw = None tb = None for c in mode: if c in "rw" and rw is None: rw = c elif c in "tb" and tb is None: tb = c else: raise ValueError("invalid mode") if rw is None: raise ValueError("must give read or write mode") if tb is None: tb = "t" if tb == "b": if encoding is not None: raise ValueError("invalid encoding argument for binary mode") if errors is not None: raise ValueError("invalid errors argument for binary mode") if newline is not None: raise ValueError("invalid newline argument for binary mode") else: if buffering == 0: raise ValueError("invalid buffering argument for text mode") if encoding is None: encoding = "ascii" if buffering == -1: buffering = self._sock.bsize raw: io.RawIOBase buf: io.BufferedIOBase if rw == "r": if size is not None: raise ValueError("size given for reading") raw = RawFileReader(self, name) if buffering == 0: return raw buf = io.BufferedReader(raw, buffering) else: if size is None: raise ValueError("size not given for writing") raw = RawFileWriter(self, name, size) if buffering == 0: return raw buf = io.BufferedWriter(raw, buffering) if tb == "t": return io.TextIOWrapper( cast(IO[bytes], buf), encoding, errors, newline, buffering == 1 ) else: return buf def find_files(self, pattern: str = "*.*") -> Iterator[tuple[str, int]]: """Find all files matching a pattern. :param pattern: Pattern to match files against. :return: An iterator on all matching files, returning file name and file size as a tuple. Accepted patterns are: - ``*.*``: to match anything (default), - ``.*``: to match files with any extension, - ``*.``: to match files with given extension, - ``.``: to match using full name. """ try: handle, name, size = self.file_find_first(pattern) except nxt.error.FileNotFoundError: return None try: yield name, size while True: try: _, name, size = self.file_find_next(handle) except nxt.error.FileNotFoundError: break yield name, size finally: self.file_close(handle) def find_modules(self, pattern: str = "*.*") -> Iterator[tuple[str, int, int, int]]: """Find all modules matching a pattern. :param pattern: Pattern to match modules against, use ``*.*`` (default) to match any module. :return: An iterator on all matching modules, returning module name, identifier, size and IO map size as a tuple. """ try: handle, mname, mid, msize, miomap_size = self.module_find_first(pattern) except nxt.error.ModuleNotFoundError: return None try: yield mname, mid, msize, miomap_size while True: try: _, mname, mid, msize, miomap_size = self.module_find_next(handle) except nxt.error.ModuleNotFoundError: break yield mname, mid, msize, miomap_size finally: self.module_close(handle) def get_motor(self, port: nxt.motor.Port) -> nxt.motor.Motor: """Return a motor object connected to one of the brick output port. :param port: Output port identifier. :return: The motor object. """ return nxt.motor.Motor(self, port) def get_sensor( self, port: nxt.sensor.Port, cls: Optional[type[nxt.sensor.Sensor]] = None, *args: Any, **kwargs: Any, ) -> nxt.sensor.Sensor: """Return a sensor object connected to one of the brick input port. :param port: Input port identifier. :param cls: Sensor class, or ``None`` to autodetect. :param args: Additional constructor positional arguments when `cls` is given. :param kwargs: Additional constructor keyword arguments when `cls` is given. :return: A sensor object. :raises nxt.sensor.digital.SearchError: When sensor can not be identified. When `cls` is not given or ``None``, try to detect the sensor type and return the correct sensor object. This only works for digital sensors with identification information. For autodetection to work, the module containing the sensor class must be imported at least once. See modules in :mod:`nxt.sensor`. """ if cls is None: if args or kwargs: raise ValueError("extra arguments with autodetect") base_sensor = nxt.sensor.digital.BaseDigitalSensor( self, port, check_compatible=False ) info = base_sensor.get_sensor_info() return nxt.sensor.digital.find_class(info)( self, port, check_compatible=False ) else: return cls(self, port, *args, **kwargs) def _cmd(self, tgram: nxt.telegram.Telegram) -> nxt.telegram.Telegram: """Send a message to the NXT brick and read reply. :param tgram: Message to send. :return: Reply message after status has been checked. """ assert tgram.reply_req with self._lock: self._sock.send(tgram.to_bytes()) reply_tgram = Telegram(opcode=tgram.opcode, pkt=self._sock.recv()) reply_tgram.check_status() return reply_tgram def _cmd_noreply(self, tgram: nxt.telegram.Telegram) -> None: """Send a message to the NXT brick with no reply. :param tgram: Message to send. """ assert not tgram.reply_req with self._lock: self._sock.send(tgram.to_bytes()) def start_program(self, name: str) -> None: """Start a program on the brick. :param name: Program file name (example: ``"myprogram.rxe"``). .. warning:: When starting or stopping a program, the NXT firmware resets every sensors and motors. """ tgram = Telegram(Opcode.DIRECT_START_PROGRAM) tgram.add_filename(name) self._cmd(tgram) def stop_program(self) -> None: """Stop the running program on the brick. :raises nxt.error.NoActiveProgramError: When no program is running. .. warning:: When starting or stopping a program, the NXT firmware resets every sensors and motors. """ tgram = Telegram(Opcode.DIRECT_STOP_PROGRAM) self._cmd(tgram) def play_sound_file(self, loop: bool, name: str) -> None: """Play a sound file on the brick. :param loop: Loop mode, play continuously. :param name: Sound file name. """ tgram = Telegram(Opcode.DIRECT_PLAY_SOUND_FILE, reply_req=False) tgram.add_bool(loop) tgram.add_filename(name) self._cmd_noreply(tgram) def play_tone(self, frequency_hz: int, duration_ms: int) -> None: """Play a tone on the brick, do not wait until finished. :param frequency_hz: Tone frequency in Hertz. :param duration_ms: Tone duration in milliseconds. This function do not wait until finished, if you want to play several notes, you may need :func:`play_tone_and_wait`. """ tgram = Telegram(Opcode.DIRECT_PLAY_TONE, reply_req=False) tgram.add_u16(frequency_hz) tgram.add_u16(duration_ms) self._cmd_noreply(tgram) def set_output_state( self, port: nxt.motor.Port, power: int, mode: nxt.motor.Mode, regulation_mode: nxt.motor.RegulationMode, turn_ratio: int, run_state: nxt.motor.RunState, tacho_limit: int, ) -> None: """Set output port state on the brick. :param port: Output port identifier. :param power: Motor speed or power level (-100 to 100). :param mode: Motor power mode. :param regulation_mode: Motor regulation mode. :param turn_ratio: Turn ratio (-100 to 100). Negative value shift power to the left motor. :param run_state: Motor run state. :param tacho_limit: Number of degrees the motor should rotate relative to the current position. .. warning:: This is a low level function, prefer to use :meth:`nxt.motor.Motor`, you can get one from :meth:`get_motor`. """ tgram = Telegram(Opcode.DIRECT_SET_OUT_STATE, reply_req=False) tgram.add_u8(port.value) tgram.add_s8(power) tgram.add_u8(mode.value) tgram.add_u8(regulation_mode.value) tgram.add_s8(turn_ratio) tgram.add_u8(run_state.value) tgram.add_u32(tacho_limit) self._cmd_noreply(tgram) def set_input_mode( self, port: nxt.sensor.Port, sensor_type: nxt.sensor.Type, sensor_mode: nxt.sensor.Mode, ) -> None: """Set input port mode on the brick. :param port: Input port identifier. :param sensor_type: Sensor type. :param sensor_mode: Sensor mode. .. warning:: This is a low level function, prefer to use a :mod:`nxt.sensor` class. """ tgram = Telegram(Opcode.DIRECT_SET_IN_MODE, reply_req=False) tgram.add_u8(port.value) tgram.add_u8(sensor_type.value) tgram.add_u8(sensor_mode.value) self._cmd_noreply(tgram) def get_output_state( self, port: nxt.motor.Port ) -> tuple[ nxt.motor.Port, int, nxt.motor.Mode, nxt.motor.RegulationMode, int, nxt.motor.RunState, int, int, int, int, ]: """Get output port state from the brick. :param port: Output port identifier. :return: A tuple with `port`, `power`, `mode`, `regulation_mode`, `turn_ratio`, `run_state`, `tacho_limit`, `tacho_count`, `block_tacho_count`, and `rotation_count`. Return value details: - **port** Output port identifier. - **power** Motor speed or power level (-100 to 100). - **mode** Motor power mode. - **regulation_mode** Motor regulation mode. - **turn_ratio** Turn ratio (-100 to 100). Negative value shift power to the left motor. - **run_state** Motor run state. - **tacho_limit** Number of degrees the motor should rotate. - **block_tacho_count** Number of degrees the motor rotated relative to the "block" start. - **rotation_count** Number of degrees the motor rotated relative to the program start. .. warning:: This is a low level function, prefer to use :meth:`nxt.motor.Motor`, you can get one from :meth:`get_motor`. """ tgram = Telegram(Opcode.DIRECT_GET_OUT_STATE) tgram.add_u8(port.value) tgram = self._cmd(tgram) port = nxt.motor.Port(tgram.parse_u8()) power = tgram.parse_s8() mode = nxt.motor.Mode(tgram.parse_u8()) regulation_mode = nxt.motor.RegulationMode(tgram.parse_u8()) turn_ratio = tgram.parse_s8() run_state = nxt.motor.RunState(tgram.parse_u8()) tacho_limit = tgram.parse_u32() tacho_count = tgram.parse_s32() block_tacho_count = tgram.parse_s32() rotation_count = tgram.parse_s32() return ( port, power, mode, regulation_mode, turn_ratio, run_state, tacho_limit, tacho_count, block_tacho_count, rotation_count, ) def get_input_values( self, port: nxt.sensor.Port ) -> tuple[ nxt.sensor.Port, bool, bool, nxt.sensor.Type, nxt.sensor.Mode, int, int, int, int, ]: """Get input port values from the brick. :param port: Input port identifier. :return: A tuple with `port`, `valid`, `calibrated`, `sensor_type`, `sensor_mode`, `raw_value`, `normalized_value`, `scaled_value`, and `calibrated_value`. `rotation_count`. Return value details: - **port** Input port identifier. - **valid** ``True`` if the value is valid, else ``False``. - **calibrated** Always ``False``, there is no calibration in NXT firmware. - **sensor_type** Sensor type. - **sensor_mode** Sensor mode. - **raw_value** Raw analog to digital converter value. - **normalized_value** Normalized value. - **scaled_value** Scaled value. - **calibrated_value** Always normalized value, there is no calibration in NXT firmware. .. warning:: This is a low level function, prefer to use a :mod:`nxt.sensor` class. """ tgram = Telegram(Opcode.DIRECT_GET_IN_VALS) tgram.add_u8(port.value) tgram = self._cmd(tgram) port = nxt.sensor.Port(tgram.parse_u8()) valid = tgram.parse_bool() calibrated = tgram.parse_bool() sensor_type = nxt.sensor.Type(tgram.parse_u8()) sensor_mode = nxt.sensor.Mode(tgram.parse_u8()) raw_value = tgram.parse_u16() normalized_value = tgram.parse_u16() scaled_value = tgram.parse_s16() calibrated_value = tgram.parse_s16() return ( port, valid, calibrated, sensor_type, sensor_mode, raw_value, normalized_value, scaled_value, calibrated_value, ) def reset_input_scaled_value(self, port: nxt.sensor.Port) -> None: """Reset scaled value for an input port on the brick. :param port: Input port identifier. This can be used to reset accumulated value for some sensor modes. .. warning:: This is a low level function, prefer to use a :mod:`nxt.sensor` class. """ tgram = Telegram(Opcode.DIRECT_RESET_IN_VAL) tgram.add_u8(port.value) self._cmd(tgram) def message_write(self, inbox: int, message: bytes) -> None: """Send a message to a brick mailbox. :param inbox: Mailbox number (0 to 19). :param message: Message to send (58 bytes maximum). """ if len(message) > 58: raise ValueError("message too long") tgram = Telegram(Opcode.DIRECT_MESSAGE_WRITE) tgram.add_u8(inbox) tgram.add_u8(len(message) + 1) tgram.add_bytes(message) tgram.add_u8(0) self._cmd(tgram) def reset_motor_position(self, port: nxt.motor.Port, relative: bool) -> None: """Reset block or program motor position for a brick output port. :param port: Output port identifier. :param relative: If ``True``, reset block position, if ``False``, reset program position. .. warning:: This is a low level function, prefer to use :meth:`nxt.motor.Motor`, you can get one from :meth:`get_motor`. """ tgram = Telegram(Opcode.DIRECT_RESET_POSITION) tgram.add_u8(port.value) tgram.add_bool(relative) self._cmd(tgram) def get_battery_level(self) -> int: """Get brick battery voltage. :return: Battery voltage in millivolt. """ tgram = Telegram(Opcode.DIRECT_GET_BATT_LVL) tgram = self._cmd(tgram) millivolts = tgram.parse_u16() return millivolts def stop_sound_playback(self) -> None: """Stop currently running sound file on the brick.""" tgram = Telegram(Opcode.DIRECT_STOP_SOUND) self._cmd(tgram) def keep_alive(self) -> int: """Reset the brick standby timer. :return: Sleep timeout in milliseconds. """ tgram = Telegram(Opcode.DIRECT_KEEP_ALIVE) tgram = self._cmd(tgram) sleep_timeout = tgram.parse_u32() return sleep_timeout def ls_get_status(self, port: nxt.sensor.Port) -> int: """Get status of last low-speed transaction to a brick input port. :param port: Input port identifier. :return: Number of bytes to read as a result of the transaction. :raises nxt.error.I2CPendingError: When transaction is still in progress. :raises nxt.error.DirectProtocolError: When there is an error on the bus. .. warning:: This is a low level function, prefer to use a :mod:`nxt.sensor` class. """ tgram = Telegram(Opcode.DIRECT_LS_GET_STATUS) tgram.add_u8(port.value) tgram = self._cmd(tgram) size = tgram.parse_u8() return size def ls_write(self, port: nxt.sensor.Port, tx_data: bytes, rx_bytes: int) -> None: """Write data to a brick input port using low speed transaction. :param port: Input port identifier. :param tx_data: Data to send. :param rx_bytes: Number of bytes to receive. Function returns immediately. Transaction status can be retrieved using :meth:`ls_get_status` and result must be read using :meth:`ls_read`. .. warning:: This is a low level function, prefer to use a :mod:`nxt.sensor` class. """ tgram = Telegram(Opcode.DIRECT_LS_WRITE) tgram.add_u8(port.value) tgram.add_u8(len(tx_data)) tgram.add_u8(rx_bytes) tgram.add_bytes(tx_data) self._cmd(tgram) def ls_read(self, port: nxt.sensor.Port) -> bytes: """Read result of low speed transaction. :param port: Input port identifier. :return: Data received. :raises nxt.error.I2CPendingError: When transaction is still in progress. :raises nxt.error.DirectProtocolError: When there is an error on the bus. The :meth:`ls_write` function must be called to initiate the transaction. .. warning:: This is a low level function, prefer to use a :mod:`nxt.sensor` class. """ tgram = Telegram(Opcode.DIRECT_LS_READ) tgram.add_u8(port.value) tgram = self._cmd(tgram) size = tgram.parse_u8() rx_data = tgram.parse_bytes(size) return rx_data def get_current_program_name(self) -> str: """Return name of program currently running on the brick. :return: Program file name :raises nxt.error.NoActiveProgramError: When no program is running. """ tgram = Telegram(Opcode.DIRECT_GET_CURR_PROGRAM) tgram = self._cmd(tgram) name = tgram.parse_filename() return name def message_read( self, remote_inbox: int, local_inbox: int, remove: bool ) -> tuple[int, bytes]: """Read a message from a brick mailbox. :param remote_inbox: Mailbox number (0 to 19). :param local_inbox: Local mailbox number, not used by brick. :param remove: Whether to remove the message from the mailbox. :return: A tuple with the local mailbox number and the read message. :raises nxt.error.EmptyMailboxError: When mailbox is empty. :raises nxt.error.NoActiveProgramError: When no program is running. """ tgram = Telegram(Opcode.DIRECT_MESSAGE_READ) tgram.add_u8(remote_inbox) tgram.add_u8(local_inbox) tgram.add_bool(remove) tgram = self._cmd(tgram) local_inbox = tgram.parse_u8() size = tgram.parse_u8() message = tgram.parse_bytes(size) return local_inbox, message def file_open_read(self, name: str) -> tuple[int, int]: """Open file for reading. :param name: File name. :return: The file handle and the file size. :raises nxt.error.FileNotFoundError: When file does not exists. .. warning:: This is a low level function, prefer to use :meth:`open_file`. """ tgram = Telegram(Opcode.SYSTEM_OPENREAD) tgram.add_filename(name) tgram = self._cmd(tgram) handle = tgram.parse_u8() size = tgram.parse_u32() return handle, size def file_open_write(self, name: str, size: int) -> int: """Open file for writing. :param name: File name. :param size: Final file size. :return: The file handle. :raises nxt.error.FileExistsError: When file already exists. :raises nxt.error.SystemProtocolError: When no space is available. .. warning:: This is a low level function, prefer to use :meth:`open_file`. """ tgram = Telegram(Opcode.SYSTEM_OPENWRITE) tgram.add_filename(name) tgram.add_u32(size) tgram = self._cmd(tgram) handle = tgram.parse_u8() return handle def file_read(self, handle: int, size: int) -> tuple[int, bytes]: """Read data from open file. :param handle: Open file handle. :param size: Number of bytes to read. :return: The file handle and the read data. .. warning:: This is a low level function, prefer to use :meth:`open_file`. """ tgram = Telegram(Opcode.SYSTEM_READ) tgram.add_u8(handle) tgram.add_u16(size) tgram = self._cmd(tgram) handle = tgram.parse_u8() size = tgram.parse_u16() data = tgram.parse_bytes(size) return handle, data def file_write(self, handle: int, data: bytes) -> tuple[int, int]: """Write data to open file. :param handle: Open file handle. :param data: Data to write. :return: The file handle and the number of bytes written. .. warning:: This is a low level function, prefer to use :meth:`open_file`. """ tgram = Telegram(Opcode.SYSTEM_WRITE) tgram.add_u8(handle) tgram.add_bytes(data) tgram = self._cmd(tgram) handle = tgram.parse_u8() size = tgram.parse_u16() return handle, size def file_close(self, handle: int) -> int: """Close open file. :param handle: Open file handle. :return: The closed file handle. .. warning:: This is a low level function, prefer to use :meth:`open_file`. """ tgram = Telegram(Opcode.SYSTEM_CLOSE) tgram.add_u8(handle) tgram = self._cmd(tgram) handle = tgram.parse_u8() return handle def file_delete(self, name: str) -> str: """Delete a file on the brick. :param name: File name. :return: The deleted file name. :raises nxt.error.FileNotFoundError: When file does not exists. """ tgram = Telegram(Opcode.SYSTEM_DELETE) tgram.add_filename(name) tgram = self._cmd(tgram) name = tgram.parse_filename() return name def file_find_first(self, pattern: str) -> tuple[int, str, int]: """Start finding files matching a pattern. :param pattern: Pattern to match files against. :return: A handle for the search, first file found name and size. :raises nxt.error.FileNotFoundError: When no file is found. .. warning:: This is a low level function, prefer to use :meth:`find_files`. """ tgram = Telegram(Opcode.SYSTEM_FINDFIRST) tgram.add_filename(pattern) tgram = self._cmd(tgram) handle = tgram.parse_u8() name = tgram.parse_filename() size = tgram.parse_u32() return handle, name, size def file_find_next(self, handle: int) -> tuple[int, str, int]: """Continue finding files. :param handle: Handle open with :meth:`file_find_first`. :return: The handle, next file found name and size. :raises nxt.error.FileNotFoundError: When no more file is found. .. warning:: This is a low level function, prefer to use :meth:`find_files`. """ tgram = Telegram(Opcode.SYSTEM_FINDNEXT) tgram.add_u8(handle) tgram = self._cmd(tgram) handle = tgram.parse_u8() name = tgram.parse_filename() size = tgram.parse_u32() return handle, name, size def get_firmware_version(self) -> tuple[tuple[int, int], tuple[int, int]]: """Get firmware version information. :return: Protocol and firmware versions, as two tuples with major and minor for each version. """ tgram = Telegram(Opcode.SYSTEM_VERSIONS) tgram = self._cmd(tgram) prot_minor = tgram.parse_u8() prot_major = tgram.parse_u8() prot_version = (prot_major, prot_minor) fw_minor = tgram.parse_u8() fw_major = tgram.parse_u8() fw_version = (fw_major, fw_minor) return prot_version, fw_version def file_open_write_linear(self, name: str, size: int) -> int: """Open file for writing, reserve a linear space. :param name: File name. :param size: Final file size. :return: The file handle. :raises nxt.error.FileExistsError: When file already exists. :raises nxt.error.SystemProtocolError: When no space is available. Linear space is required for programs, but the brick will automatically use linear mode with :meth:`file_open_write` when extension is ``.rxe``, ``.sys`` or ``.rtm``. .. warning:: This is a low level function, prefer to use :meth:`open_file`. """ tgram = Telegram(Opcode.SYSTEM_OPENWRITELINEAR) tgram.add_filename(name) tgram.add_u32(size) tgram = self._cmd(tgram) handle = tgram.parse_u8() return handle def file_open_write_data(self, name: str, size: int) -> int: """Open file for writing, using data mode. :param name: File name. :param size: Maximum file size. :return: The file handle. :raises nxt.error.FileExistsError: When file already exists. :raises nxt.error.SystemProtocolError: When no space is available. A data file can be written in small chunks, and can grow later. It can be used for data logging. .. warning:: This is a low level function, however, there is no support yet for data files in :meth:`open_file`. """ tgram = Telegram(Opcode.SYSTEM_OPENWRITEDATA) tgram.add_filename(name) tgram.add_u32(size) tgram = self._cmd(tgram) handle = tgram.parse_u8() return handle def file_open_append_data(self, name: str) -> tuple[int, int]: """Open file for appending, using data mode. :param name: File name. :return: The file handle and available size :raises nxt.error.FileNotFoundError: When file does not exists. :raises nxt.error.SystemProtocolError: When file is full or file is not a data file. The file must be a data file. The available size is the size which has not been written to last time the file was open. .. warning:: This is a low level function, however, there is no support yet for data files in :meth:`open_file`. """ tgram = Telegram(Opcode.SYSTEM_OPENAPPENDDATA) tgram.add_filename(name) tgram = self._cmd(tgram) handle = tgram.parse_u8() available_size = tgram.parse_u32() return handle, available_size def module_find_first(self, pattern: str) -> tuple[int, str, int, int, int]: """Start finding modules matching a pattern. :param pattern: Pattern to match modules against. :return: A handle for the search, first module found name, identifier, size and IO map size. :raises nxt.error.ModuleNotFoundError: When no module is found. .. warning:: This is a low level function, prefer to use :meth:`find_modules`. """ tgram = Telegram(Opcode.SYSTEM_FINDFIRSTMODULE) tgram.add_filename(pattern) tgram = self._cmd(tgram) handle = tgram.parse_u8() name = tgram.parse_filename() mod_id = tgram.parse_u32() mod_size = tgram.parse_u32() mod_iomap_size = tgram.parse_u16() return handle, name, mod_id, mod_size, mod_iomap_size def module_find_next(self, handle: int) -> tuple[int, str, int, int, int]: """Continue finding modules. :param handle: Handle open with :meth:`module_find_first`. :return: The handle, next module found name, identifier, size and IO map size. :raises nxt.error.ModuleNotFoundError: When no more module is found. .. warning:: This is a low level function, prefer to use :meth:`find_modules`. """ tgram = Telegram(Opcode.SYSTEM_FINDNEXTMODULE) tgram.add_u8(handle) tgram = self._cmd(tgram) handle = tgram.parse_u8() name = tgram.parse_filename() mod_id = tgram.parse_u32() mod_size = tgram.parse_u32() mod_iomap_size = tgram.parse_u16() return handle, name, mod_id, mod_size, mod_iomap_size def module_close(self, handle: int) -> int: """Close module search. :param handle: Open module handle. :return: The closed module handle. .. warning:: This is a low level function, prefer to use :meth:`find_modules`. """ tgram = Telegram(Opcode.SYSTEM_CLOSEMODHANDLE) tgram.add_u8(handle) tgram = self._cmd(tgram) handle = tgram.parse_u8() return handle def read_io_map(self, mod_id: int, offset: int, size: int) -> tuple[int, bytes]: """Read module IO map on the brick. :param mod_id: Module identifier. :param offset: Offset in IO map. :param size: Number of bytes to read. :return: Module identifier and read data. Module identifier can be found using :meth:`find_modules`. You need to know the structure of the module IO map. You can find it by reading the firmware source code. """ tgram = Telegram(Opcode.SYSTEM_IOMAPREAD) tgram.add_u32(mod_id) tgram.add_u16(offset) tgram.add_u16(size) tgram = self._cmd(tgram) mod_id = tgram.parse_u32() size = tgram.parse_u16() data = tgram.parse_bytes(size) return mod_id, data def write_io_map(self, mod_id: int, offset: int, data: bytes) -> tuple[int, int]: """Write module IO map on the brick. :param mod_id: Module identifier. :param offset: Offset in IO map. :param data: Data to write. :return: Module identifier and written size. Module identifier can be found using :meth:`find_modules`. You need to know the structure of the module IO map. You can find it by reading the firmware source code. """ tgram = Telegram(Opcode.SYSTEM_IOMAPWRITE) tgram.add_u32(mod_id) tgram.add_u16(offset) tgram.add_u16(len(data)) tgram.add_bytes(data) tgram = self._cmd(tgram) mod_id = tgram.parse_u32() size = tgram.parse_u16() return mod_id, size def boot(self, *, sure: bool = False) -> bytes: """Erase NXT brick firmware and go to SAM-BA boot mode. :param sure: Set to ``True`` if you are really sure. Must be a keyword parameter. :return: Brick response, should be ``"Yes\0"``. This only works on USB connection. .. danger:: This **erases the firmware** of the brick, you need to send a new firmware to use it. Sending a firmware is not supported by NXT-Python. You can use the original LEGO software or `libnxt`_ for example. Be sure to know what you are doing. .. _libnxt: https://git.ni.fr.eu.org/libnxt.git/ """ if not sure: raise ValueError("this is dangerous, please read documentation") tgram = Telegram(Opcode.SYSTEM_BOOTCMD) tgram.add_bytes(b"Let's dance: SAMBA\0") tgram = self._cmd(tgram) resp = tgram.parse_bytes() return resp def set_brick_name(self, name: str) -> None: """Set brick name. :param name: New brick name. """ tgram = Telegram(Opcode.SYSTEM_SETBRICKNAME) tgram.add_string(15, name) self._cmd(tgram) def get_device_info(self) -> tuple[str, str, tuple[int, int, int, int], int]: """Get brick information. :return: The brick name, Bluetooth address, Bluetooth signal strengths and free user flash. Bluetooth address uses this notation: ``00:16:53:xx:xx:xx``, where `xx` is the brick unique address part (`00:16:53` is the LEGO OUI used for the NXT bricks). """ tgram = Telegram(Opcode.SYSTEM_DEVICEINFO) tgram = self._cmd(tgram) name = tgram.parse_string(15) a0 = tgram.parse_u8() a1 = tgram.parse_u8() a2 = tgram.parse_u8() a3 = tgram.parse_u8() a4 = tgram.parse_u8() a5 = tgram.parse_u8() # a6 is not used, should be zero. tgram.parse_u8() address = f"{a0:02X}:{a1:02X}:{a2:02X}:{a3:02X}:{a4:02X}:{a5:02X}" signal_strengths = ( tgram.parse_u8(), tgram.parse_u8(), tgram.parse_u8(), tgram.parse_u8(), ) user_flash = tgram.parse_u32() return name, address, signal_strengths, user_flash def delete_user_flash(self) -> None: """Erase the brick user flash.""" tgram = Telegram(Opcode.SYSTEM_DELETEUSERFLASH) self._cmd(tgram) def poll_command_length(self, buf_num: int) -> tuple[int, int]: """Get number of bytes available in brick poll buffer. :param buf_num: Buffer number, 0 for USB, 1 for high speed. :return: Buffer number and number of available bytes. """ tgram = Telegram(Opcode.SYSTEM_POLLCMDLEN) tgram.add_u8(buf_num) tgram = self._cmd(tgram) buf_num = tgram.parse_u8() size = tgram.parse_u8() return buf_num, size def poll_command(self, buf_num: int, size: int) -> tuple[int, bytes]: """Get bytes from brick poll buffer. :param buf_num: Buffer number, 0 for USB, 1 for high speed. :param size: Number of bytes to read. :return: Buffer number and read bytes. """ tgram = Telegram(Opcode.SYSTEM_POLLCMD) tgram.add_u8(buf_num) tgram.add_u8(size) tgram = self._cmd(tgram) buf_num = tgram.parse_u8() size = tgram.parse_u8() command = tgram.parse_bytes(size) return buf_num, command def bluetooth_factory_reset(self) -> None: """Reset brick Bluetooth to factory settings. This only works on USB connection. """ tgram = Telegram(Opcode.SYSTEM_BTFACTORYRESET) self._cmd(tgram)