Implement v2 LPC command protocol + minor cleanup

Author: Steve-Tech

cros_ec_python/constants/LPC.py

@@ -76,4 +76,25 @@
 # EC is busy.  This covers both the EC processing a command, and the host has
 # written a new command but the EC hasn't picked it up yet.
 
-EC_LPC_STATUS_BUSY_MASK :Final = EC_LPC_STATUS_FROM_HOST | EC_LPC_STATUS_PROCESSING
+EC_LPC_STATUS_BUSY_MASK :Final = EC_LPC_STATUS_FROM_HOST | EC_LPC_STATUS_PROCESSING
+
+
+# Flags for ec_lpc_host_args.flags
+#
+# Args are from host.  Data area at EC_LPC_ADDR_HOST_PARAM contains command
+# params.
+#
+# If EC gets a command and this flag is not set, this is an old-style command.
+# Command version is 0 and params from host are at EC_LPC_ADDR_OLD_PARAM with
+# unknown length.  EC must respond with an old-style response (that is,
+# without setting EC_HOST_ARGS_FLAG_TO_HOST).
+
+EC_HOST_ARGS_FLAG_FROM_HOST: Final = 0x01
+
+# Args are from EC.  Data area at EC_LPC_ADDR_HOST_PARAM contains response.
+#
+# If EC responds to a command and this flag is not set, this is an old-style
+# response.  Command version is 0 and response data from EC is at
+# EC_LPC_ADDR_OLD_PARAM with unknown length.
+
+EC_HOST_ARGS_FLAG_TO_HOST: Final = 0x02

cros_ec_python/devices/dev.py

@@ -1,3 +1,4 @@
+import errno
 from fcntl import ioctl
 import struct
 from typing import Final
@@ -41,9 +42,9 @@ def _IORW(type: int, nr: int, size: int):
 class CrosEcDev(CrosEcClass):
     def __init__(self, fd=open("/dev/cros_ec", "wb", buffering=0), memmap_ioctl: bool = True):
         """
-
-        @param fd:
-        @param memmap_ioctl:
+        Initialise the EC using the Linux cros_ec device.
+        @param fd: Use a custom file description, opens /dev/cros_ec by default.
+        @param memmap_ioctl: Use ioctl for memmap (default), if False the READ_MEMMAP command will be used instead.
         """
         self.fd = fd
         self.memmap_ioctl = memmap_ioctl
@@ -62,6 +63,9 @@ def ec_init(self) -> None:
         pass
 
     def ec_exit(self) -> None:
+        """
+        Close the file on exit.
+        """
         if self.fd:
             self.fd.close()
 
@@ -125,8 +129,8 @@ def memmap(self, offset: Int32, num_bytes: Int32) -> bytes:
 
                 return buf[len(data): len(data) + num_bytes]
             except OSError as e:
-                if e.errno == 25:
-                    print(e)
+                if e.errno == errno.ENOTTY:
+                    warnings.warn("ioctl failed, falling back to READ_MEMMAP command", RuntimeWarning)
                     self.memmap_ioctl = False
                     return self.memmap(offset, num_bytes)
                 else:

cros_ec_python/devices/lpc.py

@@ -7,7 +7,6 @@
 from ..constants.MEMMAP import *
 from ..exceptions import ECError
 
-
 try:
     import portio
 except ImportError as e:
@@ -17,6 +16,11 @@
 
 class CrosEcLpc(CrosEcClass):
     def __init__(self, init: bool = True, address: Int32 = None):
+        """
+        Detect and initialise the EC.
+        @param init: Whether to initialise the EC on creation. Default is True.
+        @param address: Specify a custom memmap address, will be detected if not specified.
+        """
         self.address = address
         if init:
             self.ec_init()
@@ -33,6 +37,11 @@ def detect() -> bool:
 
     @staticmethod
     def find_address(*addresses) -> int | None:
+        """
+        Find the EC memory map address.
+        @param addresses: A list of addresses to check.
+        @return: The address of the EC memory map, or None if not found.
+        """
         for a in addresses:
             if res := portio.ioperm(a, EC_MEMMAP_SIZE, True):
                 if res == errno.EPERM:
@@ -48,7 +57,6 @@ def find_address(*addresses) -> int | None:
                 portio.ioperm(a, EC_MEMMAP_SIZE, False)
                 continue
 
-
     def ec_init(self) -> None:
         """
         Initialise the EC. Checks for the EC, and configures the library to speak the same version.
@@ -70,7 +78,6 @@ def ec_init(self) -> None:
             else:
                 raise OSError(f"ioperm returned {errno.errorcode[res]} ({res})")
 
-
         status = 0xFF
 
         # Read status bits, at least one should be 0
@@ -98,8 +105,75 @@ def wait_for_ec(status_addr: Int32 = EC_LPC_ADDR_HOST_CMD) -> None:
         while portio.inb(status_addr) & EC_LPC_STATUS_BUSY_MASK:
             pass
 
-    def ec_command_v2(self):
-        raise NotImplementedError
+    def ec_command_v2(self, version: UInt8, command: UInt32, outsize: UInt16, insize: UInt32, data: bytes = None,
+                      warn: bool = True):
+        """
+        Send a command to the EC and return the response. Uses the v2 command protocol over LPC. UNTESTED!
+        @param version: Command version number (often 0).
+        @param command: Command to send (EC_CMD_...).
+        @param outsize: Outgoing length in bytes.
+        @param insize: Max number of bytes to accept from the EC.
+        @param data: Outgoing data to EC.
+        @param warn: Whether to warn if the response size is not as expected. Default is True.
+        @return: Response from the EC.
+        """
+        warnings.warn("Support for v2 commands haven't been tested! Open an issue on github if it does "
+                      "or doesn't work: https://github.com/Steve-Tech/CrOS_EC_Python/issues", RuntimeWarning)
+        csum = 0
+        args = bytearray(struct.pack("BBBB", EC_HOST_ARGS_FLAG_FROM_HOST, version, outsize, csum))
+        # (flags: UInt8, command_version: UInt8, data_size: UInt8, checksum: UInt8)
+
+        # Copy data and start checksum
+        for i in range(outsize):
+            portio.outb(data[i], EC_LPC_ADDR_HOST_PARAM + i)
+            csum += data[i]
+
+        # Finish checksum
+        for i in range(len(args)):
+            csum += args[i]
+
+        args[3] = csum & 0xff
+
+        # Copy header
+        for i in range(len(args)):
+            portio.outb(args[i], EC_LPC_ADDR_HOST_ARGS + i)
+
+        # Start the command
+        portio.outb(command, EC_LPC_ADDR_HOST_CMD)
+
+        self.wait_for_ec()
+
+        # Check result
+        i = portio.inb(EC_LPC_ADDR_HOST_DATA)
+        if i:
+            raise ECError(i)
+
+        # Read back args
+        csum = 0
+        data_out = bytearray(len(args))
+        for i in range(len(data_out)):
+            data_out[i] = portio.inb(EC_LPC_ADDR_HOST_ARGS + i)
+            csum += data_out[i]
+
+        response = struct.unpack("BBBB", data_out)
+        # (flags: UInt8, command_version: UInt8, data_size: UInt8, checksum: UInt8)
+
+        if response[0] != EC_HOST_ARGS_FLAG_TO_HOST:
+            raise IOError("Invalid response!")
+
+        if response[2] != insize and warn:
+            warnings.warn(f"Expected {insize} bytes, got {response[2]} back from EC", RuntimeWarning)
+
+        # Read back data
+        data = bytearray()
+        for i in range(response[2]):
+            data.append(portio.inb(EC_LPC_ADDR_HOST_PARAM + i))
+            csum += data[i]
+
+        if response[3] != (csum & 0xff):
+            raise IOError("Checksum error!")
+
+        return bytes(data)
 
     def ec_command_v3(self, version: UInt8, command: UInt32, outsize: UInt16, insize: UInt32, data: bytes = None,
                       warn: bool = True) -> bytes:
@@ -115,6 +189,8 @@ def ec_command_v3(self, version: UInt8, command: UInt32, outsize: UInt16, insize
         """
         csum = 0
         request = bytearray(struct.pack("BBHBxH", EC_HOST_REQUEST_VERSION, csum, command, version, outsize))
+        # (struct_version: UInt8, checksum: UInt8, command: UInt16,
+        # command_version: UInt8, reserved: UInt8, data_len: UInt16)
 
         # Fail if output size is too big
         if outsize + len(request) > EC_LPC_HOST_PACKET_SIZE:
@@ -153,6 +229,7 @@ def ec_command_v3(self, version: UInt8, command: UInt32, outsize: UInt16, insize
             csum += data_out[i]
 
         response = struct.unpack("BBHHH", data_out)
+        # (struct_version: UInt8, checksum: UInt8, result: UInt16, data_len: UInt16, reserved: UInt16)
 
         if response[0] != EC_HOST_RESPONSE_VERSION:
             raise IOError("Invalid response version!")
@@ -179,7 +256,7 @@ def command(self, *args):
         """
         Stub function, will get overwritten in ec_get_cmd_version.
         """
-        raise NotImplementedError
+        raise NotImplementedError("EC doesn't support commands!")
 
     def ec_get_cmd_version(self) -> int:
         version = portio.inb(self.address + EC_MEMMAP_HOST_CMD_FLAGS)