Python library for working with the PD Buddy Sink Serial Console Configuration Interface
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

__init__.py 10.0KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334
  1. """Python bindings for PD Buddy Sink configuration"""
  2. try:
  3. # Try importing enum from the standard library
  4. import enum
  5. # Make sure Flag is available
  6. enum.Flag
  7. except (ImportError, NameError):
  8. # If something above failed, try aenum instead
  9. import aenum as enum
  10. import serial
  11. import serial.tools.list_ports
  12. class Sink:
  13. """Interface for configuring a PD Buddy Sink"""
  14. vid = 0x1209
  15. pid = 0x0001
  16. def __init__(self, sp):
  17. """Open a serial port to communicate with the PD Buddy Sink
  18. :param sp: the serial port of the device
  19. :type sp: str or `serial.tools.list_ports.ListPortInfo`
  20. """
  21. try:
  22. self._port = serial.Serial(sp, baudrate=115200)
  23. except ValueError:
  24. self._port = serial.Serial(sp.device, baudrate=115200)
  25. # Put communications in a known state, cancelling any partially-entered
  26. # command that may be sitting in the buffer.
  27. self.send_command("\x04", newline=False)
  28. def __enter__(self):
  29. return self
  30. def __exit__(self, exc_type, exc_value, traceback):
  31. self._port.close()
  32. def send_command(self, cmd, newline=True):
  33. """Send a command to the PD Buddy Sink, returning the result
  34. :param cmd: the text to send to the Sink
  35. :param newline: whether to append a ``\r\n`` to the command
  36. :type cmd: str
  37. :type newline: bool
  38. :returns: a list of zero or more bytes objects, each being one line
  39. printed as a response to the command.
  40. """
  41. # Build the command
  42. cmd = cmd.encode("utf-8")
  43. if newline:
  44. cmd += b"\r\n"
  45. # Send the command
  46. self._port.write(cmd)
  47. self._port.flush()
  48. # Read the result
  49. answer = b""
  50. while not answer.endswith(b"PDBS) "):
  51. answer += self._port.read(1)
  52. answer = answer.split(b"\r\n")
  53. # Remove the echoed command and prompt
  54. answer = answer[1:-1]
  55. return answer
  56. def close(self):
  57. """Close the serial port"""
  58. self._port.close()
  59. def help(self):
  60. """Returns the help text from the PD Buddy Sink"""
  61. return self.send_command("help")
  62. def license(self):
  63. """Returns the license text from the PD Buddy Sink"""
  64. return self.send_command("license")
  65. def erase(self):
  66. """Synchronously erases all stored configuration from flash"""
  67. self.send_command("erase")
  68. def write(self):
  69. """Synchronously writes the contents of the configuration buffer to flash"""
  70. self.send_command("write")
  71. def load(self):
  72. """Loads the current configuration from flash into the buffer
  73. :raises: KeyError
  74. """
  75. text = self.send_command("load")
  76. if len(text) > 0 and text[0].startswith(b"No configuration"):
  77. raise KeyError("no configuration")
  78. def get_cfg(self, index=None):
  79. """Reads configuration from flash
  80. :param index: optional index of configuration object in flash to read
  81. :returns: a `SinkConfig` object
  82. """
  83. if index is None:
  84. cfg = self.send_command("get_cfg")
  85. else:
  86. cfg = self.send_command("get_cfg {}".format(index))
  87. return SinkConfig.from_text(cfg)
  88. def get_tmpcfg(self):
  89. """Reads the contents of the configuration buffer
  90. :returns: a `SinkConfig` object
  91. """
  92. cfg = self.send_command("get_tmpcfg")
  93. return SinkConfig.from_text(cfg)
  94. def clear_flags(self):
  95. """Clears all the flags in the configuration buffer"""
  96. self.send_command("clear_flags")
  97. def toggle_giveback(self):
  98. """Toggles the GiveBack flag in the configuration buffer"""
  99. self.send_command("toggle_giveback")
  100. def set_v(self, mv):
  101. """Sets the voltage of the configuration buffer, in millivolts"""
  102. self.send_command("set_v {}".format(mv))
  103. def set_i(self, ma):
  104. """Sets the current of the configuration buffer, in milliamperes"""
  105. self.send_command("set_i {}".format(ma))
  106. def identify(self):
  107. """Blinks the LED quickly"""
  108. self.send_command("identify")
  109. def set_tmpcfg(self, sc):
  110. """Writes a SinkConfig object to the device's configuration buffer
  111. Note: the value of the status field is ignored; it will always be
  112. `SinkStatus.VALID`.
  113. """
  114. # Set flags
  115. self.clear_flags()
  116. if sc.flags & SinkFlags.GIVEBACK:
  117. self.toggle_giveback()
  118. # Set voltage
  119. self.set_v(sc.v)
  120. # Set current
  121. self.set_i(sc.i)
  122. @classmethod
  123. def get_devices(cls):
  124. """Get an iterable of PD Buddy Sink devices
  125. :returns: an iterable of `serial.tools.list_ports.ListPortInfo` objects
  126. """
  127. return serial.tools.list_ports.grep("{:04X}:{:04X}".format(cls.vid,
  128. cls.pid))
  129. class SinkConfig:
  130. """Python representation of a PD Buddy Sink configuration object"""
  131. def __init__(self, status=None, flags=None, v=None, i=None):
  132. """Create a SinkConfig object
  133. :param status: A `SinkStatus` value
  134. :param flags: Zero or more `SinkFlags` values
  135. :param v: Voltage in millivolts
  136. :param i: Current in milliamperes
  137. """
  138. self.status = status
  139. self.flags = flags
  140. self.v = v
  141. self.i = i
  142. def __repr__(self):
  143. s = self.__class__.__name__ + "("
  144. if self.status is not None:
  145. s += "status={}".format(self.status)
  146. if self.flags is not None:
  147. if not s.endswith("("):
  148. s += ", "
  149. s += "flags={}".format(self.flags)
  150. if self.v is not None:
  151. if not s.endswith("("):
  152. s += ", "
  153. s += "v={}".format(self.v)
  154. if self.i is not None:
  155. if not s.endswith("("):
  156. s += ", "
  157. s += "i={}".format(self.i)
  158. s += ")"
  159. return s
  160. def __str__(self):
  161. """Print the SinkStatus in the manner of the configuration shell"""
  162. s = ""
  163. if self.status is not None:
  164. s += "status: "
  165. if self.status is SinkStatus.EMPTY:
  166. s += "empty"
  167. elif self.status is SinkStatus.VALID:
  168. s += "valid"
  169. elif self.status is SinkStatus.INVALID:
  170. s += "invalid"
  171. s += "\n"
  172. if self.flags is not None:
  173. s += "flags: "
  174. if self.flags is SinkFlags.NONE:
  175. s += "(none)"
  176. else:
  177. if self.flags & SinkFlags.GIVEBACK:
  178. s += "GiveBack"
  179. s += "\n"
  180. if self.v is not None:
  181. s += "v: {:.2f} V\n".format(self.v / 1000)
  182. if self.i is not None:
  183. s += "i: {:.2f} A\n".format(self.i / 1000)
  184. # Return all but the last character of s to remove the trailing newline
  185. if s:
  186. return s[:-1]
  187. else:
  188. return "No configuration"
  189. def __eq__(self, other):
  190. if isinstance(other, self.__class__):
  191. if other.status is not self.status:
  192. return False
  193. if other.flags is not self.flags:
  194. return False
  195. if other.v != self.v:
  196. return False
  197. if other.i != self.i:
  198. return False
  199. return True
  200. return NotImplemented
  201. def __ne__(self, other):
  202. if isinstance(other, self.__class__):
  203. return not self.__eq__(other)
  204. return NotImplemented
  205. def __hash__(self):
  206. return hash(tuple(sorted(self.__dict__.items())))
  207. @classmethod
  208. def from_text(cls, text):
  209. """Creates a SinkConfig from text returned by Sink.send_command
  210. :param text: the text to load
  211. :type text: a list of bytes objects
  212. :returns: a new `SinkConfig` object.
  213. :raises: IndexError
  214. """
  215. # Assume the parameters will all be None
  216. status = None
  217. flags = None
  218. v = None
  219. i = None
  220. # Iterate over all lines of text
  221. for line in text:
  222. # If the configuration said invalid index, raise an IndexError
  223. if line.startswith(b"Invalid index"):
  224. raise IndexError("configuration index out of range")
  225. # If there is no configuration, return an empty SinkConfig
  226. elif line.startswith(b"No configuration"):
  227. return cls()
  228. # If this line is the status field
  229. elif line.startswith(b"status: "):
  230. line = line.split()[1:]
  231. if line[0] == b"empty":
  232. status = SinkStatus.EMPTY
  233. elif line[0] == b"valid":
  234. status = SinkStatus.VALID
  235. elif line[0] == b"invalid":
  236. status = SinkStatus.INVALID
  237. # If this line is the flags field
  238. elif line.startswith(b"flags: "):
  239. line = line.split()[1:]
  240. flags = SinkFlags.NONE
  241. for word in line:
  242. if word == b"(none)":
  243. # If there are no flags set, stop looking
  244. break
  245. elif word == b"GiveBack":
  246. flags |= SinkFlags.GIVEBACK
  247. # If this line is the v field
  248. elif line.startswith(b"v: "):
  249. word = line.split()[1]
  250. v = round(1000*float(word))
  251. # If this line is the i field
  252. elif line.startswith(b"i: "):
  253. word = line.split()[1]
  254. i = round(1000*float(word))
  255. # Create a new SinkConfig object with the values we just read
  256. return cls(status=status, flags=flags, v=v, i=i)
  257. class SinkStatus(enum.Enum):
  258. """Status field of a PD Buddy Sink configuration object"""
  259. EMPTY = 1
  260. VALID = 2
  261. INVALID = 3
  262. class SinkFlags(enum.Flag):
  263. """Flags field of a PD Buddy Sink configuration object"""
  264. NONE = 0
  265. GIVEBACK = enum.auto()