| Server IP : 212.252.79.165 / Your IP : 216.73.216.45 [ Web Server : Apache System : Linux 212-252-79-165.cprapid.com 5.15.0-153-generic #163-Ubuntu SMP Thu Aug 7 16:37:18 UTC 2025 x86_64 User : cehaburo ( 1001) PHP Version : 8.1.33 Disable Function : exec,passthru,shell_exec,system Domains : 48 Domains MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/internet/ |
Upload File : |
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Serial Port Protocol
"""
# dependent on pyserial ( http://pyserial.sf.net/ )
# only tested w/ 1.18 (5 Dec 2002)
from serial import PARITY_NONE # type: ignore[import]
from serial import EIGHTBITS, STOPBITS_ONE
from twisted.internet import abstract, fdesc
from twisted.internet.serialport import BaseSerialPort
class SerialPort(BaseSerialPort, abstract.FileDescriptor):
"""
A select()able serial device, acting as a transport.
"""
connected = 1
def __init__(
self,
protocol,
deviceNameOrPortNumber,
reactor,
baudrate=9600,
bytesize=EIGHTBITS,
parity=PARITY_NONE,
stopbits=STOPBITS_ONE,
timeout=0,
xonxoff=0,
rtscts=0,
):
abstract.FileDescriptor.__init__(self, reactor)
self._serial = self._serialFactory(
deviceNameOrPortNumber,
baudrate=baudrate,
bytesize=bytesize,
parity=parity,
stopbits=stopbits,
timeout=timeout,
xonxoff=xonxoff,
rtscts=rtscts,
)
self.reactor = reactor
self.flushInput()
self.flushOutput()
self.protocol = protocol
self.protocol.makeConnection(self)
self.startReading()
def fileno(self):
return self._serial.fd
def writeSomeData(self, data):
"""
Write some data to the serial device.
"""
return fdesc.writeToFD(self.fileno(), data)
def doRead(self):
"""
Some data's readable from serial device.
"""
return fdesc.readFromFD(self.fileno(), self.protocol.dataReceived)
def connectionLost(self, reason):
"""
Called when the serial port disconnects.
Will call C{connectionLost} on the protocol that is handling the
serial data.
"""
abstract.FileDescriptor.connectionLost(self, reason)
self._serial.close()
self.protocol.connectionLost(reason)