Controlling the AVR Net-IO board in Python

avr_netio.py

# DSLRlapse
# AVR Net-IO plugin
# Copyright (C) 2011 Marian Beermann
# Copyright (C) 2012 Marian Beermann
#
# 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.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

#from decimal import getcontext, Decimal
#import collections
#import time
#import os
#from PyQt4.QtCore import QThread, pyqtSlot, pyqtSignal, QCoreApplication, QTranslator, QLocale
#from PyQt4 import QtGui

#from dslrlapse import BasePlugin, CameraPlugin, Async, UnicodeTr

#from ui import Ui_controller

import telnetlib


class IO(object):
	class ConnectionError(Exception):
		pass

	class CommandError(Exception):
		pass

	class ParameterError(Exception):
		pass

	class IfConfig(object):
		def __init__(self, ip, mask, gateway):
			self.ip = ip
			self.mask = mask
			self.gateway = gateway

	@staticmethod
	def probe_host(host="192.168.0.90", port=50290):
		c = telnetlib.Telnet(host, port)
		if c.get_socket():
			return True
		else:
			return False

	def __init__(self, host="192.168.0.90", port=50290, debugging=True, timeout=5):
		"""Establishes a connection to the board"""
		self._connection = telnetlib.Telnet(host, port, timeout)

		if not self._connection.get_socket():
			raise ConnectionError

		self._outcache = dict()
		self._lcdinit = False
		self._host = host
		self._debugging = debugging

	def __del__(self):
		print "__del__"
		if hasattr(self, "_connection"):
			self._connection.close()

	def _command_std(self, command, lines=1, *args):
		command = command.upper()

		cmdstring = command
		if len(args):
			cmdstring += " " + ".".join([str(arg) for arg in args])

		if self._debugging:
			print "> %s" % cmdstring

		self._connection.write("%s\n" % (cmdstring))
		
		result = str()
		for line in range(lines):
			result += self._connection.read_until("\n")
		result.strip("\r\n")

		if self._debugging:
			print "= %s" % result

		if result.startswith("NAK"):
			if result.endswith("COMMAND"):
				raise IO.CommandError
			elif result.endswith("PARAMETER"):
				raise IO.ParameterError

		if result.startswith("ACK") or (command == "GETPORT" and result.startswith("1")):
			return True
		elif command == "GETPORT" and result == "0":
			return False
		else:
			return result

	def _command(self, command, *args):
		return self._command_std(command, 1, *args)

	def version(self):
		return self._command_std("version", 3)

	def reset(self):
		return self._command("reset")

	def get_ifconfig(self):
		"""Reads the interface configuration and returns a IO.IfConfig instance"""

		return IO.IfConfig(self._host, self._command("getmask"), self._command("getgw"))

	def set_ifconfig(self, ifconfig):
		"""Takes an IO.IfConfig instance and configures the board accordingly"""

		return self._command("setip", ifconfig.ip) and self._command("setmask", ifconfig.mask) and self._command("setgw", ifconfig.gateway)

	def get_outport(self, port_id):
		"""Read status of digital output port

		This is a very cheap operation that won't cause any network traffic in the
		majority of cases."""

		try:
			return self._outcache[port_id]
		except KeyError:
			port = self._command("getstatus")[port_id+1]
			if port == "0":
				return False
			else:
				return True

	def set_outport(self, port_id, high):
		"""Sets the output port to the given value"""

		result = self._command("setport", port_id, int(high))

		if result:
			self._outcache[port_id] = high

		return result

	def get_inport(self, port_id):
		"""Read status of digital input port"""

		return self._command("getport", port_id)

	def get_adc(self, adc_id):
		"""Read numerical value of analog input"""

		return self._command("getadc", adc_id)

	def write_lcd(self, line, text):
		"""Initializes a connected LCD (if necessary) and writes the given text
		in the given line"""

		if not self._lcdinit:
			self._lcdinit = self._command("initlcd")

		return self._command("writelcd", line, text)

	def clear_lcd(self, line=-1):
		"""Clears the LC-Display, if line is given only this line is cleared"""
		if line == -1:
			return self._command("clearlcd")
		else:
			return self._command("clearlcd", line)