arduino_serial.py

import serial, time

class _cOUTPUT(int) :
    def __new__(self) :
        return int.__new__(_cOUTPUT, 1)
    def __str__(self) :
        return "OUTPUT"
    def __repr__(self) :
        return str(self)

class _cINPUT(int) :
    def __new__(self) :
        return int.__new__(_cINPUT, 0)
    def __str__(self) :
        return "INPUT"
    def __repr__(self) :
        return str(self)

class _cHIGH(int) :
    def __new__(self) :
        return int.__new__(_cHIGH, 1)
    def __str__(self) :
        return "HIGH"
    def __repr__(self) :
        return str(self)

class _cLOW(int) :
    def __new__(self) :
        return int.__new__(_cLOW, 0)
    def __str__(self) :
        return "LOW"
    def __repr__(self) :
        return str(self)

HIGH = _cHIGH()
LOW = _cLOW()

OUTPUT = _cOUTPUT()
INPUT = _cINPUT()
    

class Arduino() :
    
    ACTIONS = {
        "ping": b'\x05',
        "set": b'\x06',
        "info": b'\x07',
        "restart": b'\x08'
    }

    RESULTS = {
        "success":   b'\x01',
        "not_found": b'\x00',
        "started":   b'\xff'
    }

    @staticmethod
    def assert_success(b) :
        assert b == Arduino.RESULTS["success"], reverse_dict(Arduino.RESULTS)[res]
    
    def __init__(self, port) :
        self._ser = serial.Serial(port, 9600)
        self.secure = False
        self.secure_connection()

    def secure_connection(self) :
        assert not self.secure, "Already secured"
        
        assert self._ser.read(1) == self.RESULTS["started"], \
                "Error during setup"
        
        self.ping()
        self.secure = True

    def restart(self) :
        self._ser.write(self.ACTIONS["restart"])
        self.assert_success(self._ser.read(1))
        
        self.secure = False
        self.secure_connection()

    def ping(self) :
        PING_STASH = b'\xab'
        
        self._ser.write(self.ACTIONS["ping"] + PING_STASH)
        res = self._ser.read(1)
        self.assert_success(res)

        assert self._ser.read(1) == PING_STASH, "Invalid echo data"

    def set_pin(self, pin, status) :
        data = pin | (128 if status else 0)
        self._ser.write(self.ACTIONS["set"]
                        + data.to_bytes(1, 'big'))
        
        res = self._ser.read(1)
        self.assert_success(res)

    def get_pin(self, pin) :
        self._ser.write(self.ACTIONS["info"] \
                + pin.to_bytes(1, 'big'))

        self.assert_success(self._ser.read(1))
        res = int.from_bytes(self._ser.read(1), 'big')

        mode = res & 1
        state = res >> 1

        return OUTPUT if mode else INPUT, HIGH if state else LOW

    def close(self) :
        self._ser.close()

    def __del__(self) :
        self.close()

def reverse_dict(dictionary) :
    return {v: k for k, v in dictionary.items()}
        

serial_connection.ino

#include <avr/io.h>
#include <avr/wdt.h>

#define reset_avr() wdt_enable(WDTO_30MS); while(1) {}
#define UNKNOWN_PIN 0xFF

uint8_t getPinMode(uint8_t pin)
{
  uint8_t bit = digitalPinToBitMask(pin);
  uint8_t port = digitalPinToPort(pin);

  // I don't see an option for mega to return this, but whatever...
  if (NOT_A_PIN == port) return UNKNOWN_PIN;

  // Is there a bit we can check?
  if (0 == bit) return UNKNOWN_PIN;

  // Is there only a single bit set?
  if (bit & bit - 1) return UNKNOWN_PIN;

  volatile uint8_t *reg, *out;
  reg = portModeRegister(port);
  out = portOutputRegister(port);

  if (*reg & bit)
    return OUTPUT;
  else if (*out & bit)
    return INPUT_PULLUP;
  else
    return INPUT;
}

// the setup function runs once when you press reset or power the board
void setup() {
  Serial.begin(9600);
  Serial.write(0xff);
}

// the loop function runs over and over again forever
void loop() {
  while(Serial.available()) {
    byte b = Serial.read();
    
    switch(b) {
      case 0x05 : {
        byte echo_data;
        while(!Serial.available()) {}
        echo_data = Serial.read();
        
        Serial.write(0x01); //Success stat byte
        Serial.write(echo_data);
        break;
      }

      case 0x06 : {
        byte data_byte;
        int pin, pinmode;
        bool stat;
        
        while(!Serial.available()) {}
        data_byte = Serial.read();

        pin = data_byte & 0b01111111;
        stat = (data_byte & 0b10000000) > 0;

        pinmode = getPinMode(pin);
        if(pinmode == 0xff) {
          Serial.write(0xff);
          break;
        }
        
        if(pinmode != OUTPUT) pinMode(pin, OUTPUT);
        digitalWrite(pin, stat ? HIGH : LOW);
        Serial.write(0x01);
        break;
      }

      case 0x07 : {
        int pin, state, mode;
        byte data_byte;

        while(!Serial.available()) {}
        pin = Serial.read();

        state = digitalRead(pin);
        mode = getPinMode(pin);

        data_byte = (state << 1) | mode;
        Serial.write(0x01);
        Serial.write(data_byte);
        break;
      }

      case 0x08 : {
        Serial.write(0x01);
        reset_avr();
      }
        
      default: {
        Serial.write(0x00); //Instruction not found byte
        break;
      }
    }
  }
}