vcr.py

Python语言编译器

import fcntl
import IOCTL
from IOCTL import *
import sys
import struct
import select
import posix
import time

DEVICE='/dev/ttyd2'

class UnixFile:
	def open(self, name, mode):
		self.fd = posix.open(name, mode)
		return self

	def read(self, len):
		return posix.read(self.fd, len)

	def write(self, data):
		dummy = posix.write(self.fd, data)

	def flush(self):
		pass

	def fileno(self):
		return self.fd

	def close(self):
		dummy = posix.close(self.fd)

def packttyargs(*args):
	if type(args) <> type(()):
		raise 'Incorrect argtype for packttyargs'
	if type(args[0]) == type(1):
		iflag, oflag, cflag, lflag, line, chars = args
	elif type(args[0]) == type(()):
		if len(args) <> 1:
			raise 'Only 1 argument expected'
		iflag, oflag, cflag, lflag, line, chars = args[0]
	elif type(args[0]) == type([]):
		if len(args) <> 1:
			raise 'Only 1 argument expected'
		[iflag, oflag, cflag, lflag, line, chars] = args[0]
	str = struct.pack('hhhhb', iflag, oflag, cflag, lflag, line)
	for c in chars:
		str = str + c
	return str

def nullttyargs():
	chars = ['\0']*IOCTL.NCCS
	return packttyargs(0, 0, 0, 0, 0, chars)

def unpackttyargs(str):
	args = str[:-IOCTL.NCCS]
	rawchars = str[-IOCTL.NCCS:]
	chars = []
	for c in rawchars:
		chars.append(c)
	iflag, oflag, cflag, lflag, line = struct.unpack('hhhhb', args)
	return (iflag, oflag, cflag, lflag, line, chars)

def initline(name):
	fp = UnixFile().open(name, 2)
	ofp = fp
	fd = fp.fileno()
	rv = fcntl.ioctl(fd, IOCTL.TCGETA, nullttyargs())
	iflag, oflag, cflag, lflag, line, chars = unpackttyargs(rv)
	iflag = iflag & ~(INPCK|ISTRIP|INLCR|IUCLC|IXON|IXOFF)
	oflag = oflag & ~OPOST
	cflag = B9600|CS8|CREAD|CLOCAL
	lflag = lflag & ~(ISIG|ICANON|ECHO|TOSTOP)
	chars[VMIN] = chr(1)
	chars[VTIME] = chr(0)
	arg = packttyargs(iflag, oflag, cflag, lflag, line, chars)
	dummy = fcntl.ioctl(fd, IOCTL.TCSETA, arg)
	return fp, ofp

#
#
error = 'VCR.error'

# Commands/replies:
COMPLETION = '\x01'
ACK  ='\x0a'
NAK  ='\x0b'

NUMBER_N = 0x30
ENTER  = '\x40'

EXP_7= '\xde'
EXP_8= '\xdf'

CL   ='\x56'
CTRL_ENABLE = EXP_8 + '\xc6'
SEARCH_DATA = EXP_8 + '\x93'
ADDR_SENSE = '\x60'

PLAY ='\x3a'
STOP ='\x3f'
EJECT='\x2a'
FF   ='\xab'
REW  ='\xac'
STILL='\x4f'
STEP_FWD ='\x2b'    # Was: '\xad'
FM_SELECT=EXP_8 + '\xc8'
FM_STILL=EXP_8 + '\xcd'
PREVIEW=EXP_7 + '\x9d'
REVIEW=EXP_7 + '\x9e'
DM_OFF=EXP_8 + '\xc9'
DM_SET=EXP_8 + '\xc4'
FWD_SHUTTLE='\xb5'
REV_SHUTTLE='\xb6'
EM_SELECT=EXP_8 + '\xc0'
N_FRAME_REC=EXP_8 + '\x92'
SEARCH_PREROLL=EXP_8 + '\x90'
EDIT_PB_STANDBY=EXP_8 + '\x96'
EDIT_PLAY=EXP_8 + '\x98'
AUTO_EDIT=EXP_7 + '\x9c'

IN_ENTRY=EXP_7 + '\x90'
IN_ENTRY_RESET=EXP_7 + '\x91'
IN_ENTRY_SET=EXP_7 + '\x98'
IN_ENTRY_INC=EXP_7 + '\x94'
IN_ENTRY_DEC=EXP_7 + '\x95'
IN_ENTRY_SENSE=EXP_7 + '\x9a'

OUT_ENTRY=EXP_7 + '\x92'
OUT_ENTRY_RESET=EXP_7 + '\x93'
OUT_ENTRY_SET=EXP_7 + '\x99'
OUT_ENTRY_INC=EXP_7 + '\x96'
OUT_ENTRY_DEC=EXP_7 + '\x98'
OUT_ENTRY_SENSE=EXP_7 + '\x9b'

MUTE_AUDIO = '\x24'
MUTE_AUDIO_OFF = '\x25'
MUTE_VIDEO = '\x26'
MUTE_VIDEO_OFF = '\x27'
MUTE_AV = EXP_7 + '\xc6'
MUTE_AV_OFF = EXP_7 + '\xc7'

DEBUG=0

class VCR:
	def __init__(self):
		self.ifp, self.ofp = initline(DEVICE)
		self.busy_cmd = None
		self.async = 0
		self.cb = None
		self.cb_arg = None

	def _check(self):
		if self.busy_cmd:
			raise error, 'Another command active: '+self.busy_cmd

	def _endlongcmd(self, name):
		if not self.async:
			self.waitready()
			return 1
		self.busy_cmd = name
		return 'VCR BUSY'

	def fileno(self):
		return self.ifp.fileno()

	def setasync(self, async):
		self.async = async

	def setcallback(self, cb, arg):
		self.setasync(1)
		self.cb = cb
		self.cb_arg = arg

	def poll(self):
		if not self.async:
			raise error, 'Can only call poll() in async mode'
		if not self.busy_cmd:
			return
		if self.testready():
			if self.cb:
				apply(self.cb, (self.cb_arg,))

	def _cmd(self, cmd):
		if DEBUG:
			print '>>>',`cmd`
		self.ofp.write(cmd)
		self.ofp.flush()

	def _waitdata(self, len, tout):
		rep = ''
		while len > 0:
			if tout == None:
				ready, d1, d2 = select.select( \
					  [self.ifp], [], [])
			else:
				ready, d1, d2 = select.select( \
					  [self.ifp], [], [], tout)
			if ready == []:
##				if rep:
##					print 'FLUSHED:', `rep`
				return None
			data = self.ifp.read(1)
			if DEBUG:
				print '<<<',`data`
			if data == NAK:
				return NAK
			rep = rep + data
			len = len - 1
		return rep

	def _reply(self, len):
		data = self._waitdata(len, 10)
		if data == None:
			raise error, 'Lost contact with VCR'
		return data

	def _getnumber(self, len):
		data = self._reply(len)
		number = 0
		for c in data:
			digit = ord(c) - NUMBER_N
			if digit < 0 or digit > 9:
				raise error, 'Non-digit in number'+`c`
			number = number*10 + digit
		return number

	def _iflush(self):
		dummy = self._waitdata(10000, 0)
##		if dummy:
##			print 'IFLUSH:', dummy

	def simplecmd(self,cmd):
		self._iflush()
		for ch in cmd:
			self._cmd(ch)
			rep = self._reply(1)
			if rep == NAK:
				return 0
			elif rep <> ACK:
				raise error, 'Unexpected reply:' + `rep`
		return 1

	def replycmd(self, cmd):
		if not self.simplecmd(cmd[:-1]):
			return 0
		self._cmd(cmd[-1])

	def _number(self, number, digits):
		if number < 0:
			raise error, 'Unexpected negative number:'+ `number`
		maxnum = pow(10, digits)
		if number > maxnum:
			raise error, 'Number too big'
		while maxnum > 1:
			number = number % maxnum
			maxnum = maxnum / 10
			digit = number / maxnum
			ok = self.simplecmd(chr(NUMBER_N + digit))
			if not ok:
				raise error, 'Error while transmitting number'

	def initvcr(self, *optarg):
		timeout = None
		if optarg <> ():
			timeout = optarg[0]
			starttime = time.time()
		self.busy_cmd = None
		self._iflush()
		while 1:
##			print 'SENDCL'
			self._cmd(CL)
			rep = self._waitdata(1, 2)
##			print `rep`
			if rep in ( None, CL, NAK ):
				if timeout:
					if time.time() - starttime > timeout:
						raise error, \
							  'No reply from VCR'
				continue
			break
		if rep <> ACK:
			raise error, 'Unexpected reply:' + `rep`
		dummy = self.simplecmd(CTRL_ENABLE)

	def waitready(self):
		rep = self._waitdata(1, None)
		if rep == None:
			raise error, 'Unexpected None reply from waitdata'
		if rep not in  (COMPLETION, ACK):
			self._iflush()
			raise error, 'Unexpected waitready reply:' + `rep`
		self.busy_cmd = None

	def testready(self):
		rep = self._waitdata(1, 0)
		if rep == None:
			return 0
		if rep not in  (COMPLETION, ACK):
			self._iflush()
			raise error, 'Unexpected waitready reply:' + `rep`
		self.busy_cmd = None
		return 1

	def play(self): return self.simplecmd(PLAY)
	def stop(self): return self.simplecmd(STOP)
	def ff(self):   return self.simplecmd(FF)
	def rew(self):  return self.simplecmd(REW)
	def eject(self):return self.simplecmd(EJECT)
	def still(self):return self.simplecmd(STILL)
	def step(self): return self.simplecmd(STEP_FWD)

	def goto(self, (h, m, s, f)):
		if not self.simplecmd(SEARCH_DATA):
			return 0
		self._number(h, 2)
		self._number(m, 2)
		self._number(s, 2)
		self._number(f, 2)
		if not self.simplecmd(ENTER):
			return 0
		return self._endlongcmd('goto')

	# XXXX TC_SENSE doesn't seem to work
	def faulty_where(self):
		self._check()
		self._cmd(TC_SENSE)
		h = self._getnumber(2)
		m = self._getnumber(2)
		s = self._getnumber(2)
		f = self._getnumber(2)
		return (h, m, s, f)

	def where(self):
		return self.addr2tc(self.sense())

	def sense(self):
		self._check()
		self._cmd(ADDR_SENSE)
		num = self._getnumber(5)
		return num

	def addr2tc(self, num):
		f = num % 25
		num = num / 25
		s = num % 60
		num = num / 60
		m = num % 60
		h = num / 60
		return (h, m, s, f)

	def tc2addr(self, (h, m, s, f)):
		return ((h*60 + m)*60 + s)*25 + f

	def fmmode(self, mode):
		self._check()
		if mode == 'off':
			arg = 0
		elif mode == 'buffer':
			arg = 1
		elif mode == 'dnr':
			arg = 2
		else:
			raise error, 'fmmode arg should be off, buffer or dnr'
		if not self.simplecmd(FM_SELECT):
			return 0
		self._number(arg, 1)
		if not self.simplecmd(ENTER):
			return 0
		return 1

	def mute(self, mode, value):
		self._check()
		if mode == 'audio':
			cmds = (MUTE_AUDIO_OFF, MUTE_AUDIO)
		elif mode == 'video':
			cmds = (MUTE_VIDEO_OFF, MUTE_VIDEO)
		elif mode == 'av':
			cmds = (MUTE_AV_OFF, MUTE_AV)
		else:
			raise error, 'mute type should be audio, video or av'
		cmd = cmds[value]
		return self.simplecmd(cmd)

	def editmode(self, mode):
		self._check()
		if mode == 'off':
			a0 = a1 = a2 = 0
		elif mode == 'format':
			a0 = 4
			a1 = 7
			a2 = 4
		elif mode == 'asmbl':
			a0 = 1
			a1 = 7
			a2 = 4
		elif mode == 'insert-video':
			a0 = 2
			a1 = 4
			a2 = 0
		else:
			raise 'editmode should be off,format,asmbl or insert-video'
		if not self.simplecmd(EM_SELECT):
			return 0
		self._number(a0, 1)
		self._number(a1, 1)
		self._number(a2, 1)
		if not self.simplecmd(ENTER):
			return 0
		return 1

	def autoedit(self):
		self._check()
		return self._endlongcmd(AUTO_EDIT)

	def nframerec(self, num):
		if not self.simplecmd(N_FRAME_REC):
			return 0
		self._number(num, 4)
		if not self.simplecmd(ENTER):
			return 0
		return self._endlongcmd('nframerec')

	def fmstill(self):
		if not self.simplecmd(FM_STILL):
			return 0
		return self._endlongcmd('fmstill')

	def preview(self):
		if not self.simplecmd(PREVIEW):
			return 0
		return self._endlongcmd('preview')

	def review(self):
		if not self.simplecmd(REVIEW):
			return 0
		return self._endlongcmd('review')

	def search_preroll(self):
		if not self.simplecmd(SEARCH_PREROLL):
			return 0
		return self._endlongcmd('search_preroll')

	def edit_pb_standby(self):
		if not self.simplecmd(EDIT_PB_STANDBY):
			return 0
		return self._endlongcmd('edit_pb_standby')

	def edit_play(self):
		if not self.simplecmd(EDIT_PLAY):
			return 0
		return self._endlongcmd('edit_play')

	def dmcontrol(self, mode):
		self._check()
		if mode == 'off':
			return self.simplecmd(DM_OFF)
		if mode == 'multi freeze':
			num = 1000
		elif mode == 'zoom freeze':
			num = 2000
		elif mode == 'digital slow':
			num = 3000
		elif mode == 'freeze':
			num = 4011
		else:
			raise error, 'unknown dmcontrol argument: ' + `mode`
		if not self.simplecmd(DM_SET):
			return 0
		self._number(num, 4)
		if not self.simplecmd(ENTER):
			return 0
		return 1

	def fwdshuttle(self, num):
		if not self.simplecmd(FWD_SHUTTLE):
			return 0
		self._number(num, 1)
		return 1

	def revshuttle(self, num):
		if not self.simplecmd(REV_SHUTTLE):
			return 0
		self._number(num, 1)
		return 1

	def getentry(self, which):
		self._check()
		if which == 'in':
			cmd = IN_ENTRY_SENSE
		elif which == 'out':
			cmd = OUT_ENTRY_SENSE
		self.replycmd(cmd)
		h = self._getnumber(2)
		m = self._getnumber(2)
		s = self._getnumber(2)
		f = self._getnumber(2)
		return (h, m, s, f)

	def inentry(self, arg):
		return self.ioentry(arg, (IN_ENTRY, IN_ENTRY_RESET, \
			  IN_ENTRY_SET, IN_ENTRY_INC, IN_ENTRY_DEC))

	def outentry(self, arg):
		return self.ioentry(arg, (OUT_ENTRY, OUT_ENTRY_RESET, \
			  OUT_ENTRY_SET, OUT_ENTRY_INC, OUT_ENTRY_DEC))

	def ioentry(self, arg, (Load, Clear, Set, Inc, Dec)):
		self._check()
		if type(arg) == type(()):
			h, m, s, f = arg
			if not self.simplecmd(Set):
				return 0
			self._number(h,2)
			self._number(m,2)
			self._number(s,2)
			self._number(f,2)
			if not self.simplecmd(ENTER):
				return 0
			return 1
		elif arg == 'reset':
			cmd = Clear
		elif arg == 'load':
			cmd = Load
		elif arg == '+':
			cmd = Inc
		elif arg == '-':
			cmd = Dec
		else:
			raise error, 'Arg should be +,-,reset,load or (h,m,s,f)'
		return self.simplecmd(cmd)

	def cancel(self):
		d = self.simplecmd(CL)
		self.busy_cmd = None