/usr/lib/python2.7/dist-packages/jnpr/junos/transport/tty_serial.py is in python-junos-eznc 2.0.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | import serial
import re
import six
from time import sleep
from datetime import datetime, timedelta
from jnpr.junos.transport.tty import Terminal
# -------------------------------------------------------------------------
# Terminal connection over SERIAL CONSOLE
# -------------------------------------------------------------------------
_PROMPT = re.compile(six.b('|').join([six.b(i) for i in Terminal._RE_PAT]))
class Serial(Terminal):
def __init__(self, port='/dev/ttyUSB0', **kvargs):
"""
:port:
the serial port, defaults to USB0 since this
:kvargs['timeout']:
this is the tty read polling timeout.
generally you should not have to tweak this.
"""
# initialize the underlying TTY device
self.port = port
self._ser = serial.Serial()
self._rx = self._ser
self._ser.port = port
self._ser.timeout = kvargs.get('timeout', self.TIMEOUT)
self._tty_name = self.port
Terminal.__init__(self, **kvargs)
# -------------------------------------------------------------------------
# I/O open close called from Terminal class
# -------------------------------------------------------------------------
def _tty_open(self):
try:
self._ser.open()
except OSError as err:
raise RuntimeError("open_failed:{0}".format(err.strerror))
self.write('\n\n\n') # hit <ENTER> a few times, yo!
def _tty_close(self):
self._ser.flush()
self._ser.close()
# -------------------------------------------------------------------------
# I/O read and write called from Terminal class
# -------------------------------------------------------------------------
def write(self, content):
""" write content + <RETURN> """
self._ser.write(six.b(content + '\n'))
self._ser.flush()
def rawwrite(self, content):
self._ser.write(content)
def read(self):
""" read a single line """
return self._ser.readline()
def read_prompt(self):
"""
reads text from the serial console (using readline) until
a match is found against the :expect: regular-expression object.
When a match is found, return a tuple(<text>,<found>) where
<text> is the complete text and <found> is the name of the
regular-expression group. If a timeout occurs, then return
the tuple(None,None).
"""
rxb = six.b('')
mark_start = datetime.now()
mark_end = mark_start + timedelta(seconds=self.EXPECT_TIMEOUT)
while datetime.now() < mark_end:
sleep(0.1) # do not remove
line = self._ser.readline()
if not line:
continue
rxb += line
found = _PROMPT.search(rxb)
if found is not None:
break # done reading
else:
# exceeded the while loop timeout
return (None, None)
return (rxb, found.lastgroup)
|