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 18KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567
  1. """Python bindings for PD Buddy Sink configuration"""
  2. from collections import namedtuple
  3. try:
  4. # Try importing enum from the standard library
  5. import enum
  6. # Make sure Flag is available
  7. enum.Flag
  8. except (ImportError, NameError, AttributeError):
  9. # If something above failed, try aenum instead
  10. import aenum as enum
  11. import serial
  12. import serial.tools.list_ports
  13. class Sink:
  14. """Interface for configuring a PD Buddy Sink"""
  15. vid = 0x1209
  16. pid = 0x9DB5
  17. def __init__(self, sp):
  18. """Open a serial port to communicate with the PD Buddy Sink
  19. :param sp: the serial port of the device
  20. :type sp: str or `serial.tools.list_ports.ListPortInfo`
  21. """
  22. try:
  23. self._port = serial.Serial(sp, baudrate=115200)
  24. except ValueError:
  25. self._port = serial.Serial(sp.device, baudrate=115200)
  26. # Put communications in a known state, cancelling any partially-entered
  27. # command that may be sitting in the buffer.
  28. self.send_command("\x04", newline=False)
  29. def __enter__(self):
  30. return self
  31. def __exit__(self, exc_type, exc_value, traceback):
  32. self._port.close()
  33. def send_command(self, cmd, newline=True):
  34. """Send a command to the PD Buddy Sink, returning the result
  35. :param cmd: the text to send to the Sink
  36. :param newline: whether to append a ``\r\n`` to the command
  37. :type cmd: str
  38. :type newline: bool
  39. :returns: a list of zero or more bytes objects, each being one line
  40. printed as a response to the command.
  41. """
  42. # Build the command
  43. cmd = cmd.encode("utf-8")
  44. if newline:
  45. cmd += b"\r\n"
  46. # Send the command
  47. self._port.write(cmd)
  48. self._port.flush()
  49. # Read the result
  50. answer = b""
  51. while not answer.endswith(b"PDBS) "):
  52. answer += self._port.read(1)
  53. answer = answer.split(b"\r\n")
  54. # Remove the echoed command and prompt
  55. answer = answer[1:-1]
  56. # Raise an exception if the command wasn't recognized
  57. if len(answer) and answer[0] == cmd.strip().split()[0] + b" ?":
  58. raise KeyError("command not found")
  59. return answer
  60. def close(self):
  61. """Close the serial port"""
  62. self._port.close()
  63. def help(self):
  64. """Returns the help text from the PD Buddy Sink"""
  65. return self.send_command("help")
  66. def license(self):
  67. """Returns the license text from the PD Buddy Sink"""
  68. return self.send_command("license")
  69. def erase(self):
  70. """Synchronously erases all stored configuration from flash"""
  71. self.send_command("erase")
  72. def write(self):
  73. """Synchronously writes the contents of the configuration buffer to flash"""
  74. self.send_command("write")
  75. def load(self):
  76. """Loads the current configuration from flash into the buffer
  77. :raises: KeyError
  78. """
  79. text = self.send_command("load")
  80. if len(text) > 0 and text[0].startswith(b"No configuration"):
  81. raise KeyError("no configuration")
  82. def get_cfg(self, index=None):
  83. """Reads configuration from flash
  84. :param index: optional index of configuration object in flash to read
  85. :returns: a `SinkConfig` object
  86. """
  87. if index is None:
  88. cfg = self.send_command("get_cfg")
  89. else:
  90. cfg = self.send_command("get_cfg {}".format(index))
  91. return SinkConfig.from_text(cfg)
  92. def get_tmpcfg(self):
  93. """Reads the contents of the configuration buffer
  94. :returns: a `SinkConfig` object
  95. """
  96. cfg = self.send_command("get_tmpcfg")
  97. return SinkConfig.from_text(cfg)
  98. def clear_flags(self):
  99. """Clears all the flags in the configuration buffer"""
  100. self.send_command("clear_flags")
  101. def toggle_giveback(self):
  102. """Toggles the GiveBack flag in the configuration buffer"""
  103. self.send_command("toggle_giveback")
  104. def set_v(self, mv):
  105. """Sets the voltage of the configuration buffer, in millivolts"""
  106. out = self.send_command("set_v {}".format(mv))
  107. # If that command gave any output, that indicates an error. Raise an
  108. # exception to make that clear.
  109. if len(out):
  110. raise ValueError(out[0])
  111. def set_i(self, ma):
  112. """Sets the current of the configuration buffer, in milliamperes"""
  113. out = self.send_command("set_i {}".format(ma))
  114. # If that command gave any output, that indicates an error. Raise an
  115. # exception to make that clear.
  116. if len(out):
  117. raise ValueError(out[0])
  118. def identify(self):
  119. """Blinks the LED quickly"""
  120. self.send_command("identify")
  121. @property
  122. def output(self):
  123. """The state of the Sink's output
  124. Raises KeyError if the ``output`` command is not available on the Sink.
  125. Raises ValueError if an invalid output is read.
  126. """
  127. value = self.send_command("output")
  128. if value[0] == b"enabled":
  129. return True
  130. elif value[0] == b"disabled":
  131. return False
  132. else:
  133. # If unexpected text is returned, raise an exception indicating a
  134. # firmware error
  135. raise ValueError("unknown output state")
  136. @output.setter
  137. def output(self, state):
  138. if state:
  139. self.send_command("output enable")
  140. else:
  141. self.send_command("output disable")
  142. def get_source_cap(self):
  143. """Gets the most recent Source_Capabilities read by the Sink"""
  144. return read_pdo_list(self.send_command("get_source_cap"))
  145. def set_tmpcfg(self, sc):
  146. """Writes a SinkConfig object to the device's configuration buffer
  147. Note: the value of the status field is ignored; it will always be
  148. `SinkStatus.VALID`.
  149. """
  150. # Set flags
  151. self.clear_flags()
  152. if sc.flags & SinkFlags.GIVEBACK:
  153. self.toggle_giveback()
  154. # Set voltage
  155. self.set_v(sc.v)
  156. # Set current
  157. self.set_i(sc.i)
  158. @classmethod
  159. def get_devices(cls):
  160. """Get an iterable of PD Buddy Sink devices
  161. :returns: an iterable of `serial.tools.list_ports.ListPortInfo` objects
  162. """
  163. return serial.tools.list_ports.grep("{:04X}:{:04X}".format(cls.vid,
  164. cls.pid))
  165. class SinkConfig(namedtuple("SinkConfig", "status flags v i")):
  166. """Python representation of a PD Buddy Sink configuration object
  167. ``status`` should be a `SinkStatus` object. ``flags`` should be zero or
  168. more `SinkFlags` values. ``v`` is the voltage in millivolts, and ``i``
  169. is the current in milliamperes. `None` is also an acceptible value for
  170. any of the fields.
  171. """
  172. __slots__ = ()
  173. def __str__(self):
  174. """Print the SinkStatus in the manner of the configuration shell"""
  175. s = ""
  176. if self.status is not None:
  177. s += "status: "
  178. if self.status is SinkStatus.EMPTY:
  179. s += "empty"
  180. elif self.status is SinkStatus.VALID:
  181. s += "valid"
  182. elif self.status is SinkStatus.INVALID:
  183. s += "invalid"
  184. s += "\n"
  185. if self.flags is not None:
  186. s += "flags: "
  187. if self.flags is SinkFlags.NONE:
  188. s += "(none)"
  189. else:
  190. if self.flags & SinkFlags.GIVEBACK:
  191. s += "GiveBack"
  192. s += "\n"
  193. if self.v is not None:
  194. s += "v: {:.2f} V\n".format(self.v / 1000.0)
  195. if self.i is not None:
  196. s += "i: {:.2f} A\n".format(self.i / 1000.0)
  197. # Return all but the last character of s to remove the trailing newline
  198. if s:
  199. return s[:-1]
  200. else:
  201. return "No configuration"
  202. @classmethod
  203. def from_text(cls, text):
  204. """Creates a SinkConfig from text returned by Sink.send_command
  205. :param text: the text to load
  206. :type text: a list of bytes objects
  207. :returns: a new `SinkConfig` object.
  208. :raises: IndexError
  209. """
  210. # Assume the parameters will all be None
  211. status = None
  212. flags = None
  213. v = None
  214. i = None
  215. # Iterate over all lines of text
  216. for line in text:
  217. # If the configuration said invalid index, raise an IndexError
  218. if line.startswith(b"Invalid index"):
  219. raise IndexError("configuration index out of range")
  220. # If there is no configuration, return an empty SinkConfig
  221. elif line.startswith(b"No configuration"):
  222. return cls(None, None, None, None)
  223. # If this line is the status field
  224. elif line.startswith(b"status: "):
  225. line = line.split()[1:]
  226. if line[0] == b"empty":
  227. status = SinkStatus.EMPTY
  228. elif line[0] == b"valid":
  229. status = SinkStatus.VALID
  230. elif line[0] == b"invalid":
  231. status = SinkStatus.INVALID
  232. # If this line is the flags field
  233. elif line.startswith(b"flags: "):
  234. line = line.split()[1:]
  235. flags = SinkFlags.NONE
  236. for word in line:
  237. if word == b"(none)":
  238. # If there are no flags set, stop looking
  239. break
  240. elif word == b"GiveBack":
  241. flags |= SinkFlags.GIVEBACK
  242. # If this line is the v field
  243. elif line.startswith(b"v: "):
  244. word = line.split()[1]
  245. v = round(1000*float(word))
  246. # If this line is the i field
  247. elif line.startswith(b"i: "):
  248. word = line.split()[1]
  249. i = round(1000*float(word))
  250. # Create a new SinkConfig object with the values we just read
  251. return cls(status=status, flags=flags, v=v, i=i)
  252. class SinkStatus(enum.Enum):
  253. """Status field of a PD Buddy Sink configuration object"""
  254. EMPTY = 1
  255. VALID = 2
  256. INVALID = 3
  257. class SinkFlags(enum.Flag):
  258. """Flags field of a PD Buddy Sink configuration object"""
  259. NONE = 0
  260. GIVEBACK = enum.auto()
  261. class UnknownPDO(namedtuple("UnknownPDO", "value")):
  262. """A PDO of an unknown type
  263. ``value`` should be a 32-bit integer representing the PDO exactly as
  264. transmitted.
  265. """
  266. __slots__ = ()
  267. pdo_type = "unknown"
  268. def __str__(self):
  269. """Print the UnknownPDO in the manner of the configuration shell"""
  270. return "{:08X}".format(self.value)
  271. class SrcFixedPDO(namedtuple("SrcFixedPDO", "dual_role_pwr usb_suspend "
  272. "unconstrained_pwr usb_comms dual_role_data peak_i v i")):
  273. """A Source Fixed PDO
  274. ``dual_role_pwr``, ``usb_suspend``, ``unconstrained_pwr``,
  275. ``usb_comms``, and ``dual_role_data`` should be booleans. ``peak_i``
  276. should be an integer in the range [0, 3]. ``v`` is the voltage in
  277. millivolts, and ``i`` is the maximum current in milliamperes.
  278. """
  279. __slots__ = ()
  280. pdo_type = "fixed"
  281. def __str__(self):
  282. """Print the SrcFixedPDO in the manner of the configuration shell"""
  283. s = self.pdo_type + "\n"
  284. if self.dual_role_pwr:
  285. s += "\tdual_role_pwr: 1\n"
  286. if self.usb_suspend:
  287. s += "\tusb_suspend: 1\n"
  288. if self.unconstrained_pwr:
  289. s += "\tunconstrained_pwr: 1\n"
  290. if self.usb_comms:
  291. s += "\tusb_comms: 1\n"
  292. if self.dual_role_data:
  293. s += "\tdual_role_data: 1\n"
  294. if self.peak_i:
  295. s += "\tpeak_i: {}\n".format(self.peak_i)
  296. s += "\tv: {:.2f} V\n".format(self.v / 1000.0)
  297. s += "\ti: {:.2f} A".format(self.i / 1000.0)
  298. return s
  299. def read_pdo(text):
  300. """Create a PDO object from partial text returned by Sink.send_command"""
  301. # First, determine the PDO type
  302. pdo_type = text[0].split(b":")[-1].strip().decode("utf-8")
  303. if pdo_type == SrcFixedPDO.pdo_type:
  304. # Set default values (n.b. there are none for v and i)
  305. dual_role_pwr = False
  306. usb_suspend = False
  307. unconstrained_pwr = False
  308. usb_comms = False
  309. dual_role_data = False
  310. peak_i = 0
  311. # Load a SrcFixedPDO
  312. for line in text[1:]:
  313. fields = line.split(b":")
  314. fields[0] = fields[0].strip()
  315. fields[1] = fields[1].strip()
  316. if fields[0] == b"dual_role_pwr":
  317. dual_role_pwr = (fields[1] == b"1")
  318. elif fields[0] == b"usb_suspend":
  319. usb_suspend = (fields[1] == b"1")
  320. elif fields[0] == b"unconstrained_pwr":
  321. unconstrained_pwr = (fields[1] == b"1")
  322. elif fields[0] == b"usb_comms":
  323. usb_comms = (fields[1] == b"1")
  324. elif fields[0] == b"dual_role_data":
  325. dual_role_data = (fields[1] == b"1")
  326. elif fields[0] == b"peak_i":
  327. peak_i = int(fields[1])
  328. elif fields[0] == b"v":
  329. v = round(1000*float(fields[1].split()[0]))
  330. elif fields[0] == b"i":
  331. i = round(1000*float(fields[1].split()[0]))
  332. # Make the SrcFixedPDO
  333. return SrcFixedPDO(
  334. dual_role_pwr=dual_role_pwr,
  335. usb_suspend=usb_suspend,
  336. unconstrained_pwr=unconstrained_pwr,
  337. usb_comms=usb_comms,
  338. dual_role_data=dual_role_data,
  339. peak_i=peak_i,
  340. v=v,
  341. i=i)
  342. elif pdo_type == "No Source_Capabilities":
  343. return None
  344. else:
  345. # Make an UnknownPDO
  346. return UnknownPDO(value=int(pdo_type, 16))
  347. def read_pdo_list(text):
  348. """Create a list of PDOs from text returned by Sink.send_command"""
  349. # Get the lines where PDOs start
  350. pdo_start_list = []
  351. for index, line in enumerate(text):
  352. if not line.startswith(b"\t"):
  353. pdo_start_list.append(index)
  354. # Append the number of lines so the last slice will work right
  355. pdo_start_list.append(len(text))
  356. # Read the PDOs
  357. pdo_list = []
  358. for start, end in zip(pdo_start_list[:-1], pdo_start_list[1:]):
  359. pdo = read_pdo(text[start:end])
  360. if pdo is not None:
  361. pdo_list.append(pdo)
  362. return pdo_list
  363. def calculate_pdp(pdo_list):
  364. """Calculate the PDP in watts of a list of PDOs
  365. The result is only guaranteed to be correct if the power supply follows
  366. the USB Power Delivery standard. Since quite a few power supplies
  367. unfortunately do not, this can really only be considered an estimate.
  368. """
  369. max_power = 0
  370. # The Source Power Rules make it so the PDP can be determined by the
  371. # highest power available from any fixed supply PDO.
  372. for pdo in pdo_list:
  373. if pdo.pdo_type == "fixed":
  374. max_power = max(max_power, pdo.v / 1000.0 * pdo.i / 1000.0)
  375. return max_power
  376. def follows_power_rules(pdo_list):
  377. """Test whether a list of PDOs follows the Power Rules for PD 2.0
  378. This function is a false-biased approximation; that is, when it returns
  379. False it is definitely correct, but when it returns True it might be
  380. incorrect.
  381. """
  382. # First, estimate the PDP assuming the rules are being followed
  383. pdp = calculate_pdp(pdo_list)
  384. # Make sure nothing exceeds the PDP
  385. for pdo in pdo_list:
  386. if pdo.pdo_type == "fixed":
  387. if pdp < pdo.v / 1000.0 * pdo.i / 1000.0:
  388. return False
  389. # TODO: in the future, there will be more types of PDO checked here
  390. # Check that the fixed supply PDOs look right
  391. seen_5v = False
  392. seen_9v = False
  393. seen_15v = False
  394. seen_20v = False
  395. seen_normative_voltages = False
  396. if pdp == 0:
  397. # No power is fine
  398. seen_normative_voltages = True
  399. elif pdp <= 15:
  400. # Below 15 W, make sure the PDP is available at 5 V.
  401. for pdo in pdo_list:
  402. if pdo.pdo_type == "fixed" and pdo.v == 5000:
  403. seen_5v = True
  404. if pdo.v / 1000.0 * pdo.i / 1000.0 != pdp:
  405. return False
  406. seen_normative_voltages = seen_5v
  407. elif pdp <= 27:
  408. # Between 15 and 27 W, make sure at least 3 A is available at 5 V and
  409. # the PDP is available at 9 V.
  410. for pdo in pdo_list:
  411. if pdo.pdo_type == "fixed" and pdo.v == 5000:
  412. seen_5v = True
  413. if pdo.i < 3000.0:
  414. return False
  415. elif pdo.pdo_type == "fixed" and pdo.v == 9000:
  416. seen_9v = True
  417. if pdo.v / 1000.0 * pdo.i / 1000.0 != pdp:
  418. return False
  419. seen_normative_voltages = seen_5v and seen_9v
  420. elif pdp <= 45:
  421. # Between 27 and 45 W, make sure at least 3 A is available at 5 and
  422. # 9 V, and the PDP is available at 15 V.
  423. for pdo in pdo_list:
  424. if pdo.pdo_type == "fixed" and pdo.v == 5000:
  425. seen_5v = True
  426. if pdo.i < 3000.0:
  427. return False
  428. elif pdo.pdo_type == "fixed" and pdo.v == 9000:
  429. seen_9v = True
  430. if pdo.i < 3000.0:
  431. return False
  432. elif pdo.pdo_type == "fixed" and pdo.v == 15000:
  433. seen_15v = True
  434. if pdo.v / 1000.0 * pdo.i / 1000.0 != pdp:
  435. return False
  436. seen_normative_voltages = seen_5v and seen_9v and seen_15v
  437. else:
  438. # Above 45 W, make sure at least 3 A is available at 5, 9, and 15 V,
  439. # and the PDP is available at 20 V.
  440. for pdo in pdo_list:
  441. if pdo.pdo_type == "fixed" and pdo.v == 5000:
  442. seen_5v = True
  443. if pdo.i < 3000.0:
  444. return False
  445. elif pdo.pdo_type == "fixed" and pdo.v == 9000:
  446. seen_9v = True
  447. if pdo.i < 3000.0:
  448. return False
  449. elif pdo.pdo_type == "fixed" and pdo.v == 15000:
  450. seen_15v = True
  451. if pdo.i < 3000.0:
  452. return False
  453. elif pdo.pdo_type == "fixed" and pdo.v == 20000:
  454. seen_20v = True
  455. if pdo.v / 1000.0 * pdo.i / 1000.0 != pdp:
  456. return False
  457. seen_normative_voltages = seen_5v and seen_9v and seen_15v and seen_20v
  458. if not seen_normative_voltages:
  459. return False
  460. # TODO: there are several things this currently doesn't test, such as
  461. # variable and battery PDOs.
  462. return True