Scribd
Upload
6 views8 pages

GXLinuxHandler: Serial Port Management

The document describes the GXLinuxHandler class, which is part of the Gurux Device Framework, providing functionality for handling serial port communication on Linux systems. It includes methods for opening and closing serial ports, configuring settings such as baud rate, data bits, parity, and stop bits, as well as handling control signals like RTS and DTR. The framework is open-source and licensed under the GNU General Public License v2.

Uploaded by

leotecnoap
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd
6 views8 pages

GXLinuxHandler: Serial Port Management

The document describes the GXLinuxHandler class, which is part of the Gurux Device Framework, providing functionality for handling serial port communication on Linux systems. It includes methods for opening and closing serial ports, configuring settings such as baud rate, data bits, parity, and stop bits, as well as handling control signals like RTS and DTR. The framework is open-source and licensed under the GNU General Public License v2.

Uploaded by

leotecnoap
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as TXT, PDF, TXT or read online on Scribd

#

# --------------------------------------------------------------------------
# Gurux Ltd
#
#
#
# Filename: $HeadURL$
#
# Version: $Revision$,
# $Date$
# $Author$
#
# Copyright (c) Gurux Ltd
#
# ---------------------------------------------------------------------------
#
# DESCRIPTION
#
# This file is a part of Gurux Device Framework.
#
# Gurux Device Framework is Open Source 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; version 2 of the License.
# Gurux Device Framework 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.
#
# More information of Gurux products: [Link]
#
# This code is licensed under the GNU General Public License v2.
# Full text may be retrieved at [Link]
# ---------------------------------------------------------------------------
# pylint: disable=import-error
import glob
import [Link]
import os
import fcntl
import array
import select
import struct
import termios
from .GXSettings import GXSettings
from .IGXNative import IGXNative

#Constant values.
_CMSPAR = 0o10000000000
_TIOCSBRK = 0x5427
_TIOCCBRK = 0x5428

_BAUDRATE_CONSTANTS = {
50: termios.B50, 75: termios.B75, 110: termios.B110, 134: termios.B134,
150: termios.B150, 200: termios.B200, 300: termios.B300,
600: termios.B600, 1200: termios.B1200, 1800: termios.B1800,
2400: termios.B2400, 4800: termios.B4800, 9600: termios.B9600,
19200: termios.B19200, 38400: termios.B38400, 57600: termios.B57600,
115200: termios.B115200, 230400: termios.B230400,
# Linux baudrates that are not included in termios module
460800: 0x1004, 500000: 0x1005, 576000: 0x1006,
921600: 0x1007, 1000000: 0x1008, 1152000: 0x1009,
1500000: 0x100A, 2000000: 0x100B, 2500000: 0x100C,
3000000: 0x100D, 3500000: 0x100E, 4000000: 0x100F,
}

_DATABITS_TO_CFLAG = {5: termios.CS5, 6: termios.CS6, 7: termios.CS7, 8:


termios.CS8}

# pylint: disable=too-many-public-methods,too-many-instance-attributes,raise-
missing-from
class GXLinuxHandler(GXSettings, IGXNative):
def __init__(self):
"""Constructor."""
GXSettings.__init__(self)
self.h = None
self.__closedR = None
self.__closedW = None

def getPortNames(self):
"""Returns available serial ports."""
tmp = [Link]('/dev/ttyS*')
[Link]([Link]('/dev/ttyUSB*'))
[Link]([Link]('/dev/ttyXRUSB*'))
[Link]([Link]('/dev/ttyACM*'))
[Link]([Link]('/dev/ttyAMA*'))
[Link]([Link]('/dev/rfcomm*'))
[Link]([Link]('/dev/ttyAP*'))
# hide non-present internal serial ports
devices = []
for device in tmp:
name = [Link](device)
if [Link]('/sys/class/tty/{}/device'.format(name)):
[Link](device)
return devices

def isOpen(self):
return self.h

def open(self, port):


#pylint: disable=bare-except,no-member
"""
Open serial port.

port: Name of serial port.


"""
[Link]()
if not port:
raise Exception("Invalid serial port name.")
self.h = [Link](port, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
(iflag, oflag, cflag, lflag, ispeed, ospeed, cc) =
[Link](self.h)
# set up raw mode / no echo / binary
cflag |= ([Link] | [Link])
lflag &= ~([Link] | [Link] | [Link] | \
[Link] | [Link] | \
[Link] | [Link]) # |[Link]
oflag &= ~([Link] | [Link] | [Link])
iflag &= ~([Link] | [Link] | [Link] | [Link])
#Baud rate
ispeed = _BAUDRATE_CONSTANTS[int([Link])]
ospeed = _BAUDRATE_CONSTANTS[int([Link])]
#Databits
cflag &= ~[Link]
if [Link] == 8:
cflag |= termios.CS8
elif [Link] == 7:
cflag |= termios.CS7
elif [Link] == 6:
cflag |= termios.CS6
elif [Link] == 5:
cflag |= termios.CS5
#Stop bits
if int([Link]) == 0:
cflag &= ~([Link])
else:
cflag |= ([Link])
# setup parity
iflag &= ~([Link] | [Link])
p = int([Link])
if p == 0: #[Link]
cflag &= ~([Link] | [Link] | _CMSPAR)
elif p == 2: #Parity.PARITY_EVEN
cflag &= ~([Link] | _CMSPAR)
cflag |= ([Link])
elif p == 1: #[Link]:
cflag &= ~_CMSPAR
cflag |= ([Link] | [Link])
elif p == 3 and _CMSPAR:#[Link]
cflag |= ([Link] | _CMSPAR | [Link])
elif p == 4 and _CMSPAR:#[Link]
cflag |= ([Link] | _CMSPAR)
cflag &= ~([Link])

#xonxoff off
iflag &= ~([Link] | [Link])
# rtscts
cflag &= ~([Link])
[Link](self.h, [Link], [iflag, oflag, cflag, lflag,
ispeed, ospeed, cc])
#Clear imput buffer.
[Link](self.h, [Link])
self.__closedR, self.__closedW = [Link]()
[Link](self.__closedR, fcntl.F_SETFL, os.O_NONBLOCK)

def close(self):
"""
Close serial port.
"""
if self.h:
[Link](self.__closedW, b'1')
[Link](self.__closedW)
[Link](self.__closedR)
self.__closedW = None
self.__closedR = None
[Link](self.h)
self.h = None

def fileno(self):
return self.h

def getBaudRate(self):
"""
Get baud rate.
"""
try:
(_, _, _, _, _, rate, _) = [Link](self.h)
except [Link] as e:
raise Exception("getBaudRate failed. " + str(e))
for k, v in _BAUDRATE_CONSTANTS.items():
if v == rate:
return k
raise Exception("Unknown baud rate: " + str(rate))

def setBaudRate(self, value):


"""
Set baud rate.
"""
try:
(iflag, oflag, cflag, lflag, ispeed, ospeed, cc) =
[Link](self.h)
except [Link] as e:
raise Exception("setBaudRate failed. " + str(e))
cflag &= ~([Link] | [Link])
cflag |= _BAUDRATE_CONSTANTS[value]
ispeed = _BAUDRATE_CONSTANTS[value]
ospeed = _BAUDRATE_CONSTANTS[value]
try:
[Link](self.h, [Link], [iflag, oflag, cflag, lflag,
ispeed, ospeed, cc])
except [Link] as e:
raise Exception("setBaudRate failed. " + str(e))

def getDataBits(self):
"""
Get data bits.
"""
try:
(_, _, cflag, _, _, _, _) = [Link](self.h)
except [Link] as e:
raise Exception("getDataBits failed. " + str(e))

cs = cflag & [Link]


for k, v in _DATABITS_TO_CFLAG.items():
if v == cs:
return k
raise Exception("Unknown data bits: " + str(cs))

def setDataBits(self, value):


"""
Set amount of data bits.
value : Amount of data bits.
"""
try:
(iflag, oflag, cflag, lflag, ispeed, ospeed, cc) =
[Link](self.h)
except [Link] as e:
raise Exception("setDataBits failed. " + str(e))
cflag &= ~[Link]
cflag |= _DATABITS_TO_CFLAG[value]
try:
[Link](self.h, [Link], [iflag, oflag, cflag, lflag,
ispeed, ospeed, cc])
except [Link] as e:
raise Exception("setDataBits failed. " + str(e))

def getParity(self):
"""Get parity.
"""
try:
(_, _, cflag, _, _, _, _) = [Link](self.h)
except [Link] as e:
raise Exception("getParity failed. " + str(e))
#[Link]
if (cflag & ([Link] | _CMSPAR | [Link])) == ([Link]
| _CMSPAR | [Link]):
return 3
#SPACE
if (cflag & ([Link] | _CMSPAR)) == ([Link] | _CMSPAR):
return 4
#ODD
if (cflag & ([Link] | [Link])) == ([Link] |
[Link]):
return 1
#EVEN
if (cflag & ([Link])) == ([Link]):
return 2
#NONE
return 0

def setParity(self, value):


"""
Set parity.
value : parity.
"""
try:
(iflag, oflag, cflag, lflag, ispeed, ospeed, cc) =
[Link](self.h)
except [Link] as e:
raise Exception("setParity failed. " + str(e))

iflag &= ~([Link] | [Link])


if [Link] == 0: #[Link]:
cflag &= ~([Link] | [Link] | _CMSPAR)
elif [Link] == 2:#[Link]:
cflag &= ~([Link] | _CMSPAR)
cflag |= ([Link])
elif [Link] == 1: #[Link]:
cflag &= ~_CMSPAR
cflag |= ([Link] | [Link])
elif [Link] == 3 and _CMSPAR: #[Link]
cflag |= ([Link] | _CMSPAR | [Link])
elif [Link] == 4 and _CMSPAR: #[Link]
cflag |= ([Link] | _CMSPAR)
cflag &= ~([Link])
# Set tty attributes
try:
[Link](self.h, [Link], [iflag, oflag, cflag, lflag,
ispeed, ospeed, cc])
except [Link] as e:
raise Exception("setParity failed. " + str(e))

def getStopBits(self):
"""
Get stop bits.
"""
try:
(_, _, cflag, _, _, _, _) = [Link](self.h)
except [Link] as e:
raise Exception("getStopBits failed. " + str(e))
if cflag & [Link] != 0:
return 1
return 0

def setStopBits(self, value):


"""
Set stop bits.
value: Amount of stop bits.
"""
try:
(iflag, oflag, cflag, lflag, ispeed, ospeed, cc) =
[Link](self.h)
except [Link] as e:
raise Exception("setStopBits failed. " + str(e))
cflag &= ~[Link]
if value == 1:
cflag |= [Link]
try:
[Link](self.h, [Link], [iflag, oflag, cflag, lflag,
ispeed, ospeed, cc])
except [Link] as e:
raise Exception("setStopBits failed. " + str(e))

def setBreakState(self, value):


"""
Set break state.
value : Is serial port in break state.
"""
if value:
[Link](self.h, _TIOCSBRK)
else:
[Link](self.h, _TIOCCBRK)

def getRtsEnable(self):
"""
Get Request To Send state.
"""
try:
(_, _, cflag, _, _, _, _) = [Link](self.h)
except [Link] as e:
raise Exception("getCtsHolding failed. " + str(e))
return (cflag & [Link]) != 0

def setRtsEnable(self, value):


"""Set Request To Send state.
value: Is RTS set.
"""
tmp = [Link]('I', 4)
if value:
[Link](self.h, 0x5416, tmp)
else:
[Link](self.h, 0x5417, tmp)

def getDtrEnable(self):
"""
Is Data Terminal ready set.
"""
try:
(_, _, cflag, _, _, _, _) = [Link](self.h)
except [Link] as e:
raise Exception("getCtsHolding failed. " + str(e))
return cflag & [Link] != 0

def setDtrEnable(self, value):


"""Is Data Terminal ready set.
value : True, if DTR is set.
"""
tmp = [Link]('I', 2)
if value:
[Link](self.h, 0x5416, tmp)
else:
[Link](self.h, 0x5417, tmp)

def getDsrHolding(self):
"""
Get Get Data Set Ready holding flag.
"""

def getBytesToRead(self):
"""
Returns amount of bytes to read.
"""
buf = [Link]('I', [0])
[Link](self.h, [Link], buf, True)
return buf[0]

def getBytesToWrite(self):
"""
Returns amount of bytes to write.
"""
buf = [Link]('I', [0])
[Link](self.h, [Link], buf, True)
return buf[0]

def read(self):
"""Read data from serial port to the buffer."""
ready, _, _ = [Link]([self.h, self.__closedR], [], [], )
if self.__closedR in ready:
return None
cnt = [Link]()
if cnt == 0:
cnt = 1
ret = [Link](self.h, cnt)
if ret is None:
return None
if isinstance(ret, int):
return bytearray([ret])
return bytearray(ret)

def write(self, data):


"""Write data to the serial port."""
ret = [Link](self.h, data)
return ret

def getCtsHolding(self):
"""Returns Clear To Send holding flag."""
try:
(_, _, cflag, _, _, _, _) = [Link](self.h)
except [Link] as e:
raise Exception("getCtsHolding failed. " + str(e))
return cflag & [Link] != 0

def getCDHolding(self):
"""Gets the state of the Carrier Detect line for the port."""

def getHandshake(self):
"""Gets the handshaking protocol for serial port transmission of data."""

def setHandshake(self, value):


# pylint: disable=attribute-defined-outside-init
"""Sets the handshaking protocol for serial port transmission of data."""

You might also like