rs485.py 3.2 KB

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