Added get_source_cap and PDO support

Now there are namedtuples for PDO types, and two functions to help with
reading PDOs from the Sink.  They're all tested, and our test coverage
is now at 98%(!).

README.rst

@@ -12,6 +12,7 @@ Features
 -  SinkConfig objects can be manipulated locally and written to the
    device with one method call
 -  Allows control of whether or not the output is enabled
+-  Provides a Pythonic interface for reading advertised PDOs
 
 Examples
 --------

pdbuddy/__init__.py

@@ -180,6 +180,9 @@ class Sink:
         else:
             self.send_command("output disable")
 
+    def get_source_cap(self):
+        """Gets the most recent Source_Capabilities read by the Sink"""
+        return read_pdo_list(self.send_command("get_source_cap"))
 
     def set_tmpcfg(self, sc):
         """Writes a SinkConfig object to the device's configuration buffer
@@ -320,3 +323,128 @@ class SinkFlags(enum.Flag):
     """Flags field of a PD Buddy Sink configuration object"""
     NONE = 0
     GIVEBACK = enum.auto()
+
+
+class UnknownPDO(namedtuple("UnknownPDO", "value")):
+    """A PDO of an unknown type
+
+    ``value`` should be a 32-bit integer representing the PDO exactly as
+    transmitted.
+    """
+    __slots__ = ()
+
+    pdo_type = "unknown"
+
+    def __str__(self):
+        """Print the UnknownPDO in the manner of the configuration shell"""
+        return "{:08X}".format(self.value)
+
+
+class SrcFixedPDO(namedtuple("SrcFixedPDO", "dual_role_pwr usb_suspend "
+        "unconstrained_pwr usb_comms dual_role_data peak_i v i")):
+    """A Source Fixed PDO
+
+    ``dual_role_pwr``, ``usb_suspend``, ``unconstrained_pwr``,
+    ``usb_comms``, and ``dual_role_data`` should be booleans.  ``peak_i``
+    should be an integer in the range [0, 3].  ``v`` is the voltage in
+    millivolts, and ``i`` is the maximum current in milliamperes.
+    """
+    __slots__ = ()
+
+    pdo_type = "fixed"
+
+    def __str__(self):
+        """Print the SrcFixedPDO in the manner of the configuration shell"""
+        s = self.pdo_type + "\n"
+
+        if self.dual_role_pwr:
+            s += "\tdual_role_pwr: 1\n"
+
+        if self.usb_suspend:
+            s += "\tusb_suspend: 1\n"
+
+        if self.unconstrained_pwr:
+            s += "\tunconstrained_pwr: 1\n"
+
+        if self.usb_comms:
+            s += "\tusb_comms: 1\n"
+
+        if self.dual_role_data:
+            s += "\tdual_role_data: 1\n"
+
+        if self.peak_i:
+            s += "\tpeak_i: {}\n".format(self.peak_i)
+
+        s += "\tv: {:.2f} V\n".format(self.v / 1000)
+        s += "\ti: {:.2f} A".format(self.i / 1000)
+
+        return s
+
+
+def read_pdo(text):
+    """Create a PDO object from partial text returned by Sink.send_command"""
+    # First, determine the PDO type
+    pdo_type = text[0].split(b":")[-1].strip().decode("utf-8")
+
+    if pdo_type == SrcFixedPDO.pdo_type:
+        # Set default values (n.b. there are none for v and i)
+        dual_role_pwr = False
+        usb_suspend = False
+        unconstrained_pwr = False
+        usb_comms = False
+        dual_role_data = False
+        peak_i = 0
+
+        # Load a SrcFixedPDO
+        for line in text[1:]:
+            fields = line.split(b":")
+            fields[0] = fields[0].strip()
+            fields[1] = fields[1].strip()
+            if fields[0] == b"dual_role_pwr":
+                dual_role_pwr = (fields[1] == b"1")
+            elif fields[0] == b"usb_suspend":
+                usb_suspend = (fields[1] == b"1")
+            elif fields[0] == b"unconstrained_pwr":
+                unconstrained_pwr = (fields[1] == b"1")
+            elif fields[0] == b"usb_comms":
+                usb_comms = (fields[1] == b"1")
+            elif fields[0] == b"dual_role_data":
+                dual_role_data = (fields[1] == b"1")
+            elif fields[0] == b"peak_i":
+                peak_i = int(fields[1])
+            elif fields[0] == b"v":
+                v = round(1000*float(fields[1].split()[0]))
+            elif fields[0] == b"i":
+                i = round(1000*float(fields[1].split()[0]))
+
+        # Make the SrcFixedPDO
+        return SrcFixedPDO(
+                dual_role_pwr=dual_role_pwr,
+                usb_suspend=usb_suspend,
+                unconstrained_pwr=unconstrained_pwr,
+                usb_comms=usb_comms,
+                dual_role_data=dual_role_data,
+                peak_i=peak_i,
+                v=v,
+                i=i)
+    else:
+        # Make an UnknownPDO
+        return UnknownPDO(value=int(pdo_type, 16))
+
+
+def read_pdo_list(text):
+    """Create a list of PDOs from text returned by Sink.send_command"""
+    # Get the lines where PDOs start
+    pdo_start_list = []
+    for index, line in enumerate(text):
+        if not line.startswith(b"\t"):
+            pdo_start_list.append(index)
+    # Append the number of lines so the last slice will work right
+    pdo_start_list.append(len(text))
+
+    # Read the PDOs
+    pdo_list = []
+    for start, end in zip(pdo_start_list[:-1], pdo_start_list[1:]):
+        pdo_list.append(read_pdo(text[start:end]))
+
+    return pdo_list

test_pdbuddy/__init__.py

@@ -240,3 +240,113 @@ class SinkConfigTestCase(unittest.TestCase):
                 b"v: 15.00 V",
                 b"i: 3.00 A"])
         self.assertEqual(ft_valid, self.obj_valid)
+
+
+class UnknownPDOTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.obj_zero = pdbuddy.UnknownPDO(value=0x00000000)
+        self.obj_notzero = pdbuddy.UnknownPDO(value=0xFFFFFFFF)
+
+    def test_str_zero(self):
+        self.assertEqual(str(self.obj_zero), "00000000")
+
+    def test_str_notzero(self):
+        self.assertEqual(str(self.obj_notzero), "FFFFFFFF")
+
+
+class SrcFixedPDOTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.obj_everything = pdbuddy.SrcFixedPDO(True, True, True, True, True,
+                3, 20000, 5000)
+        self.obj_minimal = pdbuddy.SrcFixedPDO(False, False, False, False,
+                False, 0, 5000, 1500)
+
+    def test_str_everything(self):
+        self.assertEqual(str(self.obj_everything),
+                "fixed\n\tdual_role_pwr: 1\n\tusb_suspend: 1\n"
+                "\tunconstrained_pwr: 1\n\tusb_comms: 1\n\tdual_role_data: 1\n"
+                "\tpeak_i: 3\n\tv: 20.00 V\n\ti: 5.00 A")
+
+    def test_str_minimal(self):
+        self.assertEqual(str(self.obj_minimal),
+                "fixed\n\tv: 5.00 V\n\ti: 1.50 A")
+
+
+class ReadPDOTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.src_fixed_everything = pdbuddy.SrcFixedPDO(True, True, True, True,
+                True, 3, 20000, 5000)
+        self.src_fixed_minimal = pdbuddy.SrcFixedPDO(False, False, False,
+                False, False, 0, 5000, 1500)
+        self.unknown_zero = pdbuddy.UnknownPDO(value=0x00000000)
+        self.unknown_notzero = pdbuddy.UnknownPDO(value=0xFFFFFFFF)
+
+    def test_read_src_fixed_everything(self):
+        rp_src_fixed_everything = pdbuddy.read_pdo([b"PDO 1: fixed",
+                b"\tdual_role_pwr: 1",
+                b"\tusb_suspend: 1",
+                b"\tunconstrained_pwr: 1",
+                b"\tusb_comms: 1",
+                b"\tdual_role_data: 1",
+                b"\tpeak_i: 3",
+                b"\tv: 20.00 V",
+                b"\ti: 5.00 A"])
+        self.assertEqual(self.src_fixed_everything, rp_src_fixed_everything)
+
+    def test_read_src_fixed_minimal(self):
+        rp_src_fixed_minimal = pdbuddy.read_pdo([b"PDO 1: fixed",
+                b"\tv: 5.00 V",
+                b"\ti: 1.50 A"])
+        self.assertEqual(self.src_fixed_minimal, rp_src_fixed_minimal)
+
+    def test_read_src_fixed_minimal_no_index(self):
+        rp_src_fixed_minimal = pdbuddy.read_pdo([b"fixed",
+                b"\tv: 5.00 V",
+                b"\ti: 1.50 A"])
+        self.assertEqual(self.src_fixed_minimal, rp_src_fixed_minimal)
+
+    def test_read_unknown_zero(self):
+        rp_unknown_zero = pdbuddy.read_pdo([b"PDO 1: 00000000"])
+        self.assertEqual(self.unknown_zero, rp_unknown_zero)
+
+    def test_read_unknown_notzero(self):
+        rp_unknown_notzero = pdbuddy.read_pdo([b"PDO 1: FFFFFFFF"])
+        self.assertEqual(self.unknown_notzero, rp_unknown_notzero)
+
+
+class ReadPDOListTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.src_fixed_everything = pdbuddy.SrcFixedPDO(True, True, True, True,
+                True, 3, 20000, 5000)
+        self.src_fixed_minimal = pdbuddy.SrcFixedPDO(False, False, False,
+                False, False, 0, 5000, 1500)
+        self.unknown_zero = pdbuddy.UnknownPDO(value=0x00000000)
+        self.unknown_notzero = pdbuddy.UnknownPDO(value=0xFFFFFFFF)
+
+    def test_read_pdo_list(self):
+        # It's not a legal list for USB Power Delivery, but it works fine for
+        # testing this code
+        text = [b"PDO 1: fixed",
+                b"\tdual_role_pwr: 1",
+                b"\tusb_suspend: 1",
+                b"\tunconstrained_pwr: 1",
+                b"\tusb_comms: 1",
+                b"\tdual_role_data: 1",
+                b"\tpeak_i: 3",
+                b"\tv: 20.00 V",
+                b"\ti: 5.00 A",
+                b"PDO 2: fixed",
+                b"\tv: 5.00 V",
+                b"\ti: 1.50 A",
+                b"PDO 3: 00000000",
+                b"PDO 4: FFFFFFFF"]
+        pdo_list = pdbuddy.read_pdo_list(text)
+
+        self.assertEqual(pdo_list[0], self.src_fixed_everything)
+        self.assertEqual(pdo_list[1], self.src_fixed_minimal)
+        self.assertEqual(pdo_list[2], self.unknown_zero)
+        self.assertEqual(pdo_list[3], self.unknown_notzero)