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
|