tools/pyalpha/serial/rs485.py

   1 #!/usr/bin/env python
   2
   3 # RS485 support
   4 #
   5 # This file is part of pySerial. https://github.com/pyserial/pyserial
   6 # (C) 2015 Chris Liechti <cliechti@gmx.net>
   7 #
   8 # SPDX-License-Identifier:    BSD-3-Clause
   9
  10 """\
  11 The settings for RS485 are stored in a dedicated object that can be applied to
  12 serial ports (where supported).
  13 NOTE: Some implementations may only support a subset of the settings.
  14 """
  15
  16 import time
  17 import serial
  18
  19
  20 class RS485Settings(object):
  21     def __init__(
  22             self,
  23             rts_level_for_tx=True,
  24             rts_level_for_rx=False,
  25             loopback=False,
  26             delay_before_tx=None,
  27             delay_before_rx=None):
  28         self.rts_level_for_tx = rts_level_for_tx
  29         self.rts_level_for_rx = rts_level_for_rx
  30         self.loopback = loopback
  31         self.delay_before_tx = delay_before_tx
  32         self.delay_before_rx = delay_before_rx
  33
  34
  35 class RS485(serial.Serial):
  36     """\
  37     A subclass that replaces the write method with one that toggles RTS
  38     according to the RS485 settings.
  39
  40     NOTE: This may work unreliably on some serial ports (control signals not
  41           synchronized or delayed compared to data). Using delays may be
  42           unreliable (varying times, larger than expected) as the OS may not
  43           support very fine grained delays (no smaller than in the order of
  44           tens of milliseconds).
  45
  46     NOTE: Some implementations support this natively. Better performance
  47           can be expected when the native version is used.
  48
  49     NOTE: The loopback property is ignored by this implementation. The actual
  50           behavior depends on the used hardware.
  51
  52     Usage:
  53
  54         ser = RS485(...)
  55         ser.rs485_mode = RS485Settings(...)
  56         ser.write(b'hello')
  57     """
  58
  59     def __init__(self, *args, **kwargs):
  60         super(RS485, self).__init__(*args, **kwargs)
  61         self._alternate_rs485_settings = None
  62
  63     def write(self, b):
  64         """Write to port, controlling RTS before and after transmitting."""
  65         if self._alternate_rs485_settings is not None:
  66             # apply level for TX and optional delay
  67             self.setRTS(self._alternate_rs485_settings.rts_level_for_tx)
  68             if self._alternate_rs485_settings.delay_before_tx is not None:
  69                 time.sleep(self._alternate_rs485_settings.delay_before_tx)
  70             # write and wait for data to be written
  71             super(RS485, self).write(b)
  72             super(RS485, self).flush()
  73             # optional delay and apply level for RX
  74             if self._alternate_rs485_settings.delay_before_rx is not None:
  75                 time.sleep(self._alternate_rs485_settings.delay_before_rx)
  76             self.setRTS(self._alternate_rs485_settings.rts_level_for_rx)
  77         else:
  78             super(RS485, self).write(b)
  79
  80     # redirect where the property stores the settings so that underlying Serial
  81     # instance does not see them
  82     @property
  83     def rs485_mode(self):
  84         """\
  85         Enable RS485 mode and apply new settings, set to None to disable.
  86         See serial.rs485.RS485Settings for more info about the value.
  87         """
  88         return self._alternate_rs485_settings
  89
  90     @rs485_mode.setter
  91     def rs485_mode(self, rs485_settings):
  92         self._alternate_rs485_settings = rs485_settings