Python library for working with the PD Buddy Sink Serial Console Configuration Interface
Du kannst nicht mehr als 25 Themen auswählen Themen müssen mit entweder einem Buchstaben oder einer Ziffer beginnen. Sie können Bindestriche („-“) enthalten und bis zu 35 Zeichen lang sein.

__init__.py 20KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605
  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. class TypeCVirtualPDO(namedtuple("TypeCVirtualPDO", "i")):
  300. """A Type-C Current Virtual PDO
  301. ``i`` is the advertised current in milliamperes.
  302. """
  303. __slots__ = ()
  304. pdo_type = "typec_virtual"
  305. def __str__(self):
  306. """Print the TypeCVirtualPDO in the manner of the configuration shell"""
  307. s = self.pdo_type + "\n"
  308. s += "\ti: {:.2f} A".format(self.i / 1000.0)
  309. return s
  310. def read_pdo(text):
  311. """Create a PDO object from partial text returned by Sink.send_command"""
  312. # First, determine the PDO type
  313. pdo_type = text[0].split(b":")[-1].strip().decode("utf-8")
  314. if pdo_type == SrcFixedPDO.pdo_type:
  315. # Set default values (n.b. there are none for v and i)
  316. dual_role_pwr = False
  317. usb_suspend = False
  318. unconstrained_pwr = False
  319. usb_comms = False
  320. dual_role_data = False
  321. peak_i = 0
  322. # Load a SrcFixedPDO
  323. for line in text[1:]:
  324. fields = line.split(b":")
  325. fields[0] = fields[0].strip()
  326. fields[1] = fields[1].strip()
  327. if fields[0] == b"dual_role_pwr":
  328. dual_role_pwr = (fields[1] == b"1")
  329. elif fields[0] == b"usb_suspend":
  330. usb_suspend = (fields[1] == b"1")
  331. elif fields[0] == b"unconstrained_pwr":
  332. unconstrained_pwr = (fields[1] == b"1")
  333. elif fields[0] == b"usb_comms":
  334. usb_comms = (fields[1] == b"1")
  335. elif fields[0] == b"dual_role_data":
  336. dual_role_data = (fields[1] == b"1")
  337. elif fields[0] == b"peak_i":
  338. peak_i = int(fields[1])
  339. elif fields[0] == b"v":
  340. v = round(1000*float(fields[1].split()[0]))
  341. elif fields[0] == b"i":
  342. i = round(1000*float(fields[1].split()[0]))
  343. # Make the SrcFixedPDO
  344. return SrcFixedPDO(
  345. dual_role_pwr=dual_role_pwr,
  346. usb_suspend=usb_suspend,
  347. unconstrained_pwr=unconstrained_pwr,
  348. usb_comms=usb_comms,
  349. dual_role_data=dual_role_data,
  350. peak_i=peak_i,
  351. v=v,
  352. i=i)
  353. elif pdo_type == TypeCVirtualPDO.pdo_type:
  354. # Load a TypeCVirtualPDO
  355. for line in text[1:]:
  356. fields = line.split(b":")
  357. fields[0] = fields[0].strip()
  358. fields[1] = fields[1].strip()
  359. if fields[0] == b"i":
  360. i = round(1000*float(fields[1].split()[0]))
  361. # Make the TypeCVirtualPDO
  362. return TypeCVirtualPDO(i=i)
  363. elif pdo_type == "No Source_Capabilities":
  364. return None
  365. else:
  366. # Make an UnknownPDO
  367. return UnknownPDO(value=int(pdo_type, 16))
  368. def read_pdo_list(text):
  369. """Create a list of PDOs from text returned by Sink.send_command"""
  370. # Get the lines where PDOs start
  371. pdo_start_list = []
  372. for index, line in enumerate(text):
  373. if not line.startswith(b"\t"):
  374. pdo_start_list.append(index)
  375. # Append the number of lines so the last slice will work right
  376. pdo_start_list.append(len(text))
  377. # Read the PDOs
  378. pdo_list = []
  379. for start, end in zip(pdo_start_list[:-1], pdo_start_list[1:]):
  380. pdo = read_pdo(text[start:end])
  381. if pdo is not None:
  382. pdo_list.append(pdo)
  383. return pdo_list
  384. def calculate_pdp(pdo_list):
  385. """Calculate the PDP in watts of a list of PDOs
  386. The result is only guaranteed to be correct if the power supply follows
  387. the USB Power Delivery standard. Since quite a few power supplies
  388. unfortunately do not, this can really only be considered an estimate.
  389. """
  390. max_power = 0
  391. # The Source Power Rules make it so the PDP can be determined by the
  392. # highest power available from any fixed supply PDO.
  393. for pdo in pdo_list:
  394. if pdo.pdo_type == "fixed":
  395. max_power = max(max_power, pdo.v / 1000.0 * pdo.i / 1000.0)
  396. elif pdo.pdo_type == "typec_virtual":
  397. max_power = max(max_power, 5.0 * pdo.i / 1000.0)
  398. return max_power
  399. def follows_power_rules(pdo_list):
  400. """Test whether a list of PDOs follows the Power Rules for PD 2.0
  401. This function is a false-biased approximation; that is, when it returns
  402. False it is definitely correct, but when it returns True it might be
  403. incorrect.
  404. """
  405. # First, estimate the PDP assuming the rules are being followed
  406. pdp = calculate_pdp(pdo_list)
  407. # If there's a typec_virtual PDO, there's no Power Delivery so the Power
  408. # Rules cannot be violated. In truth they're not really being followed
  409. # either since they only apply to Power Delivery, but returning True here
  410. # seems like the safer option.
  411. if pdo_list and pdo_list[0].pdo_type == "typec_virtual":
  412. return True
  413. # Make sure nothing exceeds the PDP
  414. for pdo in pdo_list:
  415. if pdo.pdo_type == "fixed":
  416. if pdp < pdo.v / 1000.0 * pdo.i / 1000.0:
  417. return False
  418. # TODO: in the future, there will be more types of PDO checked here
  419. # Check that the fixed supply PDOs look right
  420. seen_5v = False
  421. seen_9v = False
  422. seen_15v = False
  423. seen_20v = False
  424. seen_normative_voltages = False
  425. if pdp == 0:
  426. # No power is fine
  427. seen_normative_voltages = True
  428. elif pdp <= 15:
  429. # Below 15 W, make sure the PDP is available at 5 V.
  430. for pdo in pdo_list:
  431. if pdo.pdo_type == "fixed" and pdo.v == 5000:
  432. seen_5v = True
  433. if pdo.v / 1000.0 * pdo.i / 1000.0 != pdp:
  434. return False
  435. seen_normative_voltages = seen_5v
  436. elif pdp <= 27:
  437. # Between 15 and 27 W, make sure at least 3 A is available at 5 V and
  438. # the PDP is available at 9 V.
  439. for pdo in pdo_list:
  440. if pdo.pdo_type == "fixed" and pdo.v == 5000:
  441. seen_5v = True
  442. if pdo.i < 3000.0:
  443. return False
  444. elif pdo.pdo_type == "fixed" and pdo.v == 9000:
  445. seen_9v = True
  446. if pdo.v / 1000.0 * pdo.i / 1000.0 != pdp:
  447. return False
  448. seen_normative_voltages = seen_5v and seen_9v
  449. elif pdp <= 45:
  450. # Between 27 and 45 W, make sure at least 3 A is available at 5 and
  451. # 9 V, and the PDP is available at 15 V.
  452. for pdo in pdo_list:
  453. if pdo.pdo_type == "fixed" and pdo.v == 5000:
  454. seen_5v = True
  455. if pdo.i < 3000.0:
  456. return False
  457. elif pdo.pdo_type == "fixed" and pdo.v == 9000:
  458. seen_9v = True
  459. if pdo.i < 3000.0:
  460. return False
  461. elif pdo.pdo_type == "fixed" and pdo.v == 15000:
  462. seen_15v = True
  463. if pdo.v / 1000.0 * pdo.i / 1000.0 != pdp:
  464. return False
  465. seen_normative_voltages = seen_5v and seen_9v and seen_15v
  466. else:
  467. # Above 45 W, make sure at least 3 A is available at 5, 9, and 15 V,
  468. # and the PDP is available at 20 V.
  469. for pdo in pdo_list:
  470. if pdo.pdo_type == "fixed" and pdo.v == 5000:
  471. seen_5v = True
  472. if pdo.i < 3000.0:
  473. return False
  474. elif pdo.pdo_type == "fixed" and pdo.v == 9000:
  475. seen_9v = True
  476. if pdo.i < 3000.0:
  477. return False
  478. elif pdo.pdo_type == "fixed" and pdo.v == 15000:
  479. seen_15v = True
  480. if pdo.i < 3000.0:
  481. return False
  482. elif pdo.pdo_type == "fixed" and pdo.v == 20000:
  483. seen_20v = True
  484. if pdo.v / 1000.0 * pdo.i / 1000.0 != pdp:
  485. return False
  486. seen_normative_voltages = seen_5v and seen_9v and seen_15v and seen_20v
  487. if not seen_normative_voltages:
  488. return False
  489. # TODO: there are several things this currently doesn't test, such as
  490. # variable and battery PDOs.
  491. return True