<commit_before>import serial
import time
from hedgehog.periphery import SERIAL as SERIAL_PORT
from hedgehog.protocol.errors import FailedCommandError
from . import HardwareAdapter, POWER

# commands:
IO_STATE = 0x10
ANALOG_REQ = 0x20
DIGITAL_REQ = 0x30
MOTOR = 0x40
SERVO = 0x50
SERIAL = 0x60
# replies:
OK = 0x80
UNKNOWN_OPCODE = 0x81
INVALID_OPCODE = 0x82
INVALID_PORT = 0x83
INVALID_IO = 0x84
INVALID_MODE = 0x85
INVALID_FLAGS = 0x86
INVALID_VALUE = 0x87
ANALOG_REP = 0xA1
DIGITAL_REP = 0xB1
SERIAL_UPDATE = 0xE1
# number analog and digital ports together, servo and motor ports separately
# special analog ports:
BATTERY_VOLTAGE = 0x80
# special digital ports (output only):
LED1 = 0x90
LED2 = 0x91
BUZZER = 0x92
# serial ports:
SPI1 = 0x00
SPI2 = 0x01

_replies = {
    OK,
    ANALOG_REP,
    DIGITAL_REP,
}
_errors = {
    UNKNOWN_OPCODE,
    INVALID_OPCODE,
    INVALID_PORT,
    INVALID_IO,
    INVALID_MODE,
    INVALID_FLAGS,
    INVALID_VALUE,
}
_cmd_lengths = {
    OK: 1,
    UNKNOWN_OPCODE: 1,
    INVALID_OPCODE: 1,
    INVALID_PORT: 1,
    INVALID_IO: 1,
    INVALID_MODE: 1,
    INVALID_FLAGS: 1,
    INVALID_VALUE: 1,
    ANALOG_REP: 4,
    DIGITAL_REP: 3,
}

class SerialHardwareAdapter(HardwareAdapter):
    def __init__(self, motor_state_update_cb=None):
        super().__init__(motor_state_update_cb=motor_state_update_cb)
        self.serial = serial.Serial(
            port=SERIAL_PORT,
            baudrate=115200,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_EVEN,
            stopbits=serial.STOPBITS_ONE,
            timeout=5,
            xonxoff=False,
            rtscts=False,
            writeTimeout=None,
            dsrdtr=False,
            interCharTimeout=None,
        )

    def command(self, cmd, reply_code=OK):
        def read_command():
            cmd = self.serial.read()
            if cmd[0] in _cmd_lengths:
                length = _cmd_lengths[cmd[0]]
                if length > 1:
                    cmd += self.serial.read(length - 1)
            else:
                cmd += self.serial.read()
                cmd += self.serial.read(cmd[1])
            return list(cmd)

        self.serial.write(bytes(cmd))
        reply = read_command()
        while reply[0] not in _replies | _errors:
            # TODO do something with the update
            reply = read_command()
<commit_msg>do a microcontroller reset in server start<commit_after>import serial
import time
from hedgehog.periphery import GPIO, NRST_PIN, SERIAL as SERIAL_PORT
from hedgehog.protocol.errors import FailedCommandError
from . import HardwareAdapter, POWER

# commands:
IO_STATE = 0x10
ANALOG_REQ = 0x20
DIGITAL_REQ = 0x30
MOTOR = 0x40
SERVO = 0x50
SERIAL = 0x60
# replies:
OK = 0x80
UNKNOWN_OPCODE = 0x81
INVALID_OPCODE = 0x82
INVALID_PORT = 0x83
INVALID_IO = 0x84
INVALID_MODE = 0x85
INVALID_FLAGS = 0x86
INVALID_VALUE = 0x87
ANALOG_REP = 0xA1
DIGITAL_REP = 0xB1
SERIAL_UPDATE = 0xE1
# number analog and digital ports together, servo and motor ports separately
# special analog ports:
BATTERY_VOLTAGE = 0x80
# special digital ports (output only):
LED1 = 0x90
LED2 = 0x91
BUZZER = 0x92
# serial ports:
SPI1 = 0x00
SPI2 = 0x01

_replies = {
    OK,
    ANALOG_REP,
    DIGITAL_REP,
}
_errors = {
    UNKNOWN_OPCODE,
    INVALID_OPCODE,
    INVALID_PORT,
    INVALID_IO,
    INVALID_MODE,
    INVALID_FLAGS,
    INVALID_VALUE,
}
_cmd_lengths = {
    OK: 1,
    UNKNOWN_OPCODE: 1,
    INVALID_OPCODE: 1,
    INVALID_PORT: 1,
    INVALID_IO: 1,
    INVALID_MODE: 1,
    INVALID_FLAGS: 1,
    INVALID_VALUE: 1,
    ANALOG_REP: 4,
    DIGITAL_REP: 3,
}

class SerialHardwareAdapter(HardwareAdapter):
    def __init__(self, motor_state_update_cb=None):
        super().__init__(motor_state_update_cb=motor_state_update_cb)
        self.serial = serial.Serial(
            port=SERIAL_PORT,
            baudrate=115200,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_EVEN,
            stopbits=serial.STOPBITS_ONE,
            timeout=5,
            xonxoff=False,
            rtscts=False,
            writeTimeout=None,
            dsrdtr=False,
            interCharTimeout=None,
        )

        self.reset = GPIO(NRST_PIN)
        self.reset.setup(GPIO.OUT)

        self.reset.set(False)
        time.sleep(0.2)
        self.reset.set(True)
        time.sleep(0.8)

        self.serial.flushInput()
        self.serial.flushOutput()


    def command(self, cmd, reply_code=OK):
        def read_command():
            cmd = self.serial.read()
            if cmd[0] in _cmd_lengths:
                length = _cmd_lengths[cmd[0]]
                if length > 1:
                    cmd += self.serial.read(length - 1)
            else:
                cmd += self.serial.read()
                cmd += self.serial.read(cmd[1])
            return list(cmd)

        self.serial.write(bytes(cmd))
        reply = read_command()
        while reply[0] not in _replies | _errors:
            # TODO do something with the update
            reply = read_command()

{'fraction_non_alphanumeric': 0.08349861166203887, 'fraction_numerical': 0.03768345894486315, 'mean_word_length': 2.3868368032236402, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2867931', 'n_tokens_mistral': 1898, 'n_tokens_neox': 1695, 'n_words': 383}