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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
#
# PROFIBUS DP - Communication Processor PHY access library
#
# Copyright (c) 2016 Michael Buesch <m@bues.ch>
#
# Licensed under the terms of the GNU General Public License version 2,
# or (at your option) any later version.
#
from pyprofibus.phy import *
from pyprofibus.fdl import FdlTelegram
import sys
import time
import binascii
try:
import serial
except ImportError as e:
if "PyPy" in sys.version and\
sys.version_info[0] == 2:
# We are on PyPy2.
# Try to import CPython2's serial.
import glob
sys.path.extend(glob.glob("/usr/lib/python2*/*-packages/"))
import serial
else:
raise e
class CpPhySerial(CpPhy):
"""pyserial based PROFIBUS CP PHYsical layer
"""
def __init__(self, port, debug = False):
super(CpPhySerial, self).__init__(debug = debug)
self.__port = port
self.__serial = None
self.setConfig()
def close(self):
if self.__serial:
try:
self.__serial.close()
except serial.SerialException as e:
pass
self.__serial = None
self.__rxBuf = bytearray()
super(CpPhySerial, self).close()
def __discard(self):
if self.__serial:
self.__serial.flushInput()
self.__serial.flushOutput()
# Poll for received packet.
# timeout => In seconds. 0 = none, Negative = unlimited.
def poll(self, timeout = 0):
ret, rxBuf, s, size = None, self.__rxBuf, self.__serial, -1
getSize = FdlTelegram.getSizeFromRaw
timeoutStamp = time.clock() + timeout
try:
while True:
if len(rxBuf) < 1:
rxBuf += s.read(1)
elif len(rxBuf) < 3:
try:
size = getSize(rxBuf)
readLen = size
except ProfibusError:
readLen = 3
rxBuf += s.read(readLen - len(rxBuf))
elif len(rxBuf) >= 3:
try:
size = getSize(rxBuf)
except ProfibusError:
rxBuf = bytearray()
self.__discard()
raise PhyError("PHY-serial: "
"Failed to get received "
"telegram size:\n"
"Invalid telegram format.")
if len(rxBuf) < size:
rxBuf += s.read(size - len(rxBuf))
if len(rxBuf) == size:
ret, rxBuf = rxBuf, bytearray()
break
if timeout >= 0 and\
time.clock() >= timeoutStamp:
break
except serial.SerialException as e:
rxBuf = bytearray()
self.__discard()
raise PhyError("PHY-serial: Failed to receive "
"telegram:\n" + str(e))
finally:
self.__rxBuf = rxBuf
if self.debug and ret:
print("PHY-serial: RX %s" %\
binascii.b2a_hex(ret).decode())
return ret
def setConfig(self, baudrate = CpPhy.BAUD_19200):
self.close()
try:
self.__serial = serial.Serial(
port = self.__port,
baudrate = baudrate,
bytesize = 8,
parity = serial.PARITY_EVEN,
stopbits = serial.STOPBITS_ONE,
timeout = 0,
xonxoff = False,
rtscts = False,
dsrdtr = False)
self.__rxBuf = bytearray()
except serial.SerialException as e:
raise PhyError("Failed to set CP-PHY "
"configuration:\n" + str(e))
self.__setConfigPiLC(baudrate)
def __setConfigPiLC(self, baudrate):
"""Reconfigure the PiLC HAT, if available.
"""
try:
import libpilc.raspi_hat_conf as raspi_hat_conf
except ImportError as e:
return
if not raspi_hat_conf.PilcConf.havePilcHat():
return
try:
conf = raspi_hat_conf.PilcConf()
conf.setBaudrate(baudrate / 1000.0)
except raspi_hat_conf.PilcConf.Error as e:
raise PhyError("Failed to configure PiLC HAT:\n%s" %\
str(e))
def profibusSend_SDN(self, telegramData):
try:
telegramData = bytearray(telegramData)
if self.debug:
print("PHY-serial: TX %s" %\
binascii.b2a_hex(telegramData).decode())
self.__serial.write(telegramData)
except serial.SerialException as e:
raise PhyError("PHY-serial: Failed to transmit "
"telegram:\n" + str(e))
profibusSend_SRD = profibusSend_SDN
|
