First draft for a computer side python implementation

Sebastian 2015-06-26 14:53:58 +02:00
parent 3e4abc5db3
commit 699424ed7c
3 changed files with 136 additions and 2 deletions

.gitignore vendored
View File

@ -3,3 +3,6 @@

View File

@ -57,9 +57,9 @@ Transport Layer
A device is allowed to send data after the bus has been idle for at least *48 bit times* (aka. 5ms).
After this period the sender is allowed to send a single frame.
The frame format is simply the first byte is the length of the entire frame in bytes (`= N + 3`),
The frame format is simple: The first byte is the length of the entire frame in bytes (`= N + 3`),
followed by N bytes of payload.
The remaining 2 bytes of the are a 16bit CRC checksum over the frames payload.
The remaining 2 bytes are a 16bit CRC checksum over the frames payload.
If a received frame has timed out or if its CRC sum does not match its payload,
any receiving devices should assume a collision and drop the frame.
A frame can be assumed as timed out if it was not fully received yet and no
@ -73,6 +73,7 @@ If the read byte does not match the written byte, a collision occurred.
In this case the sender has to wait for random back off interval until attempting a retransmit.
The back off interval is calculated as `(8 * bit_time) * (priority + rand(1,5))`,
where `bit_time = 1 / 9600 s` and priority is the frames priority.
After the backoff interval the bus has to be idle for *48 bit times* before retransmit.
Device receiving on the bus are not necessarily able to detect collisions,
unless they show up as timing error at their UART.
Therefore receiving devices should rely only on the length and the CRC sum of frame to detect collisions.

View File

@ -0,0 +1,130 @@
#!/usr/bin/env python2
import serial
import random
import time
from crcmod.predefined import PredefinedCrc
# Timings in bit times
class SSS7Bus(object):
def __init__(self, port, baudrate=9600):
self._bit_time = 1.0 / baudrate
self._serial = serial.Serial(port, baudrate, timeout=self._bit_time * TIMEOUT)
self._buffer = []
def _debug(self, msg):
print "[SSS7Bus] %s" % msg
def _read_frame(self, clear_channel=False):
if clear_channel :
self._serial.timeout = self._bit_time * CLEARCHANNEL
self._debug("Checking if bus is idle for %f seconds" % (self._bit_time * CLEARCHANNEL))
self._debug("Trying to read a frame from the bus")
self._serial.timeout = timeout=self._bit_time * TIMEOUT
length_byte =
if len(length_byte) == 0:
if clear_channel:
self._debug("Bus was idle")
self._debug("Timeout reading frame length")
if clear_channel:
self._debug("Bus was not idle, trying to read frame")
length = ord(length_byte) - 1 # we read the size length byte already
self._serial.timeout = timeout=self._bit_time * TIMEOUT
self._debug("Trying to read remaing %d bytes of frame" % length)
data =
# if read returns less then length, no new byte has been received for timeout seconds
if len(data) != length:
self._debug("Timeout reading frame payload")
crc = data[-2:]
payload = data[:-2]
crc16 = PredefinedCrc('crc16')
real_crc = crc16.digest()
if real_crc != crc:
self._debug("Wrong crc for frame payload")
return self.ERR_CRC_ERRROR
self._debug("Sucessfully read rest of frame")
return self.ERR_SUCCESS
def _flush_input_buffer(self):
while self._serial.inWaiting() > 0:
self._debug("Flushing input buffer")
def _can_send(self):
return self._read_frame(True) == self.ERR_TIMEOUT_ON_LENGTH
def _send_byte(self, byte):
return byte ==
def _send_frame(self, frame):
while not self._can_send():
for byte in frame:
result = self._send_byte(byte)
if not result:
self._debug("Sending frame failed with collision")
return False
self._debug("Successfully send frame")
return True
def send_message(self, msg, priority=5):
crc16 = PredefinedCrc('crc16')
msg_crc = crc16.digest()
frame = chr(len(msg) + 3) + msg + msg_crc
result = self._send_frame(frame)
while not result:
backoff = 8 * self._bit_time * (priority + random.randint(1,5))
self._debug("Collision occured backing off %f" % backoff)
result = self._send_frame(frame)
def read_message(self):
while len(self._buffer) == 0:
return self._buffer.pop()