Improve LPC/devportio performance

Author: Steve-Tech

cros_ec_python/devices/lpc.py

@@ -7,30 +7,24 @@
 from ..constants.MEMMAP import *
 from ..exceptions import ECError
 
-try:
-    import portio
-    _import_portio_error = None
-except ImportError as e:
-    _import_portio_error = e
-    warnings.warn(f"Failed to import portio: {e}, using /dev/port instead.", ImportWarning)
-    from ..utils import devportio as portio
+from ..ioports import PortIO
 
 
 class CrosEcLpc(CrosEcClass):
     """
     Class to interact with the EC using the LPC interface.
     """
 
-    def __init__(self, init: bool = True, address: Int32 = None, portio_warn: bool = True):
+    def __init__(self, init: bool = True, address: Int32 = None, portio: PortIO = PortIO()):
         """
         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.
-        :param portio_warn: Whether to warn if portio couldn't be imported. Default is True.
+        :param portio: PortIO object to use. Default is auto-detected.
         """
-        if portio_warn and _import_portio_error is not None:
-            warnings.warn(f"Failed to import portio: {_import_portio_error}, using /dev/port instead.",
-                          RuntimeWarning)
+
+        self.portio: PortIO = portio
+        """PortIO object to use."""
 
         self.address: Int32 = address
         """The address of the EC memory map."""
@@ -49,7 +43,7 @@ def detect() -> bool:
                     return True
 
     @staticmethod
-    def find_address(*addresses) -> int | None:
+    def find_address(*addresses, portio: PortIO = PortIO()) -> int | None:
         """
         Find the EC memory map address.
         :param addresses: A list of addresses to check.
@@ -77,15 +71,15 @@ def ec_init(self) -> None:
         """
         # Find memmap address
         if self.address is None:
-            self.address = self.find_address(EC_LPC_ADDR_MEMMAP, EC_LPC_ADDR_MEMMAP_FWAMD)
+            self.address = self.find_address(EC_LPC_ADDR_MEMMAP, EC_LPC_ADDR_MEMMAP_FWAMD, portio=self.portio)
             # find_address will leave ioperm enabled for the memmap
             if self.address is None:
                 raise OSError("Could not find EC!")
 
         # Request I/O permissions
-        if (res := portio.ioperm(EC_LPC_ADDR_HOST_DATA, EC_MEMMAP_SIZE, True)) or \
-                (res := portio.ioperm(EC_LPC_ADDR_HOST_CMD, EC_MEMMAP_SIZE, True)) or \
-                (res := portio.ioperm(EC_LPC_ADDR_HOST_PACKET, EC_LPC_HOST_PACKET_SIZE, True)):
+        if (res := self.portio.ioperm(EC_LPC_ADDR_HOST_DATA, EC_MEMMAP_SIZE, True)) or \
+                (res := self.portio.ioperm(EC_LPC_ADDR_HOST_CMD, EC_MEMMAP_SIZE, True)) or \
+                (res := self.portio.ioperm(EC_LPC_ADDR_HOST_PACKET, EC_LPC_HOST_PACKET_SIZE, True)):
             if res == errno.EPERM:
                 raise PermissionError("Permission denied. Try running as root.")
             else:
@@ -94,28 +88,27 @@ def ec_init(self) -> None:
         status = 0xFF
 
         # Read status bits, at least one should be 0
-        status &= portio.inb(EC_LPC_ADDR_HOST_CMD)
-        status &= portio.inb(EC_LPC_ADDR_HOST_DATA)
+        status &= self.portio.inb(EC_LPC_ADDR_HOST_CMD)
+        status &= self.portio.inb(EC_LPC_ADDR_HOST_DATA)
 
         if status == 0xFF:
             raise OSError("No EC detected. Invalid status.")
 
         # Check for 'EC' in memory map
-        if portio.inw(self.address + EC_MEMMAP_ID) != int.from_bytes(b'EC', "little"):
+        if self.portio.inw(self.address + EC_MEMMAP_ID) != int.from_bytes(b'EC', "little"):
             raise OSError("Invalid EC signature.")
 
         self.ec_get_cmd_version()
 
     def ec_exit(self) -> None:
         pass
 
-    @staticmethod
-    def wait_for_ec(status_addr: Int32 = EC_LPC_ADDR_HOST_CMD) -> None:
+    def wait_for_ec(self, status_addr: Int32 = EC_LPC_ADDR_HOST_CMD) -> None:
         """
         Wait for the EC to be ready after sending a command.
         :param status_addr: The status register to read.
         """
-        while portio.inb(status_addr) & EC_LPC_STATUS_BUSY_MASK:
+        while self.portio.inb(status_addr) & EC_LPC_STATUS_BUSY_MASK:
             pass
 
     def ec_command_v2(self, version: UInt8, command: UInt32, outsize: UInt16, insize: UInt32, data: bytes = None,
@@ -138,7 +131,7 @@ def ec_command_v2(self, version: UInt8, command: UInt32, outsize: UInt16, insize
 
         # Copy data and start checksum
         for i in range(outsize):
-            portio.outb(data[i], EC_LPC_ADDR_HOST_PARAM + i)
+            self.portio.outb(data[i], EC_LPC_ADDR_HOST_PARAM + i)
             csum += data[i]
 
         # Finish checksum
@@ -149,23 +142,23 @@ def ec_command_v2(self, version: UInt8, command: UInt32, outsize: UInt16, insize
 
         # Copy header
         for i in range(len(args)):
-            portio.outb(args[i], EC_LPC_ADDR_HOST_ARGS + i)
+            self.portio.outb(args[i], EC_LPC_ADDR_HOST_ARGS + i)
 
         # Start the command
-        portio.outb(command, EC_LPC_ADDR_HOST_CMD)
+        self.portio.outb(command, EC_LPC_ADDR_HOST_CMD)
 
         self.wait_for_ec()
 
         # Check result
-        i = portio.inb(EC_LPC_ADDR_HOST_DATA)
+        i = self.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)
+            data_out[i] = self.portio.inb(EC_LPC_ADDR_HOST_ARGS + i)
             csum += data_out[i]
 
         response = struct.unpack("BBBB", data_out)
@@ -180,7 +173,7 @@ def ec_command_v2(self, version: UInt8, command: UInt32, outsize: UInt16, insize
         # Read back data
         data = bytearray()
         for i in range(response[2]):
-            data.append(portio.inb(EC_LPC_ADDR_HOST_PARAM + i))
+            data.append(self.portio.inb(EC_LPC_ADDR_HOST_PARAM + i))
             csum += data[i]
 
         if response[3] != (csum & 0xff):
@@ -211,7 +204,7 @@ def ec_command_v3(self, version: UInt8, command: UInt32, outsize: UInt16, insize
 
         # Copy data and start checksum
         for i in range(outsize):
-            portio.outb(data[i], EC_LPC_ADDR_HOST_PACKET + len(request) + i)
+            self.portio.outb(data[i], EC_LPC_ADDR_HOST_PACKET + len(request) + i)
             csum += data[i]
 
         # Finish checksum
@@ -222,23 +215,23 @@ def ec_command_v3(self, version: UInt8, command: UInt32, outsize: UInt16, insize
 
         # Copy header
         for i in range(len(request)):
-            portio.outb(request[i], EC_LPC_ADDR_HOST_PACKET + i)
+            self.portio.outb(request[i], EC_LPC_ADDR_HOST_PACKET + i)
 
         # Start the command
-        portio.outb(EC_COMMAND_PROTOCOL_3, EC_LPC_ADDR_HOST_CMD)
+        self.portio.outb(EC_COMMAND_PROTOCOL_3, EC_LPC_ADDR_HOST_CMD)
 
         self.wait_for_ec()
 
         # Check result
-        i = portio.inb(EC_LPC_ADDR_HOST_DATA)
+        i = self.portio.inb(EC_LPC_ADDR_HOST_DATA)
         if i:
             raise ECError(i)
 
         # Read back response and start checksum
         csum = 0
         data_out = bytearray(1 + 1 + 2 + 2 + 2)
         for i in range(len(data_out)):
-            data_out[i] = portio.inb(EC_LPC_ADDR_HOST_PACKET + i)
+            data_out[i] = self.portio.inb(EC_LPC_ADDR_HOST_PACKET + i)
             csum += data_out[i]
 
         response = struct.unpack("BBHHH", data_out)
@@ -257,7 +250,7 @@ def ec_command_v3(self, version: UInt8, command: UInt32, outsize: UInt16, insize
         # Read back data
         data = bytearray()
         for i in range(response[3]):
-            data.append(portio.inb(EC_LPC_ADDR_HOST_PACKET + len(data_out) + i))
+            data.append(self.portio.inb(EC_LPC_ADDR_HOST_PACKET + len(data_out) + i))
             csum += data[i]
 
         if csum & 0xff:
@@ -276,7 +269,7 @@ def ec_get_cmd_version(self) -> int:
         Find the version of the EC command protocol.
         :return: The version of the EC command protocol.
         """
-        version = portio.inb(self.address + EC_MEMMAP_HOST_CMD_FLAGS)
+        version = self.portio.inb(self.address + EC_MEMMAP_HOST_CMD_FLAGS)
 
         if version & EC_HOST_CMD_FLAG_VERSION_3:
             self.command = self.ec_command_v3
@@ -298,5 +291,5 @@ def memmap(self, offset: Int32, num_bytes: Int32) -> bytes:
         """
         data = bytearray()
         for i in range(num_bytes):
-            data.append(portio.inb(self.address + offset + i))
+            data.append(self.portio.inb(self.address + offset + i))
         return bytes(data)

cros_ec_python/ioports/__init__.py

@@ -0,0 +1,13 @@
+"""
+This module is used to import the correct portio module based on the OS.
+"""
+
+import os
+
+if os.name == "posix":
+    try:
+        from .x86portio import IoPortIo as PortIO
+    except ImportError:
+        from .devportio import DevPortIO as PortIO
+else:
+    raise Exception("Unsupported OS")

cros_ec_python/ioports/baseportio.py

@@ -0,0 +1,87 @@
+"""
+This file includes the base class for Port I/O backends.
+
+It doesn't do anything on its own, but it is used as a base for all port I/O backends to inherit from.
+
+See `cros_ec_python.io` for a few examples of a classes that inherits from this class.
+"""
+
+import abc
+
+
+class PortIOClass(metaclass=abc.ABCMeta):
+    """
+    Base class for port I/O backends to inherit from.
+    """
+
+    @abc.abstractmethod
+    def outb(self, data: int, port: int) -> None:
+        """
+        Write a byte (8 bit) to the specified port.
+        :param data: Byte to write.
+        :param port: Port to write to.
+        """
+        pass
+
+    @abc.abstractmethod
+    def outw(self, data: int, port: int) -> None:
+        """
+        Write a word (16 bit) to the specified port.
+        :param data: Word to write.
+        :param port: Port to write to.
+        """
+        pass
+
+    @abc.abstractmethod
+    def outl(self, data: int, port: int) -> None:
+        """
+        Write a long (32 bit) to the specified port.
+        :param data: Long to write.
+        :param port: Port to write to.
+        """
+        pass
+
+    @abc.abstractmethod
+    def inb(self, port: int) -> int:
+        """
+        Read a byte (8 bit) from the specified port.
+        :param port: Port to read from.
+        :return: Byte read.
+        """
+        pass
+
+    @abc.abstractmethod
+    def inw(self, port: int) -> int:
+        """
+        Read a word (16 bit) from the specified port.
+        :param port: Port to read from.
+        :return: Word read.
+        """
+        pass
+
+    @abc.abstractmethod
+    def inl(self, port: int) -> int:
+        """
+        Read a long (32 bit) from the specified port.
+        :param port: Port to read from.
+        :return: Long read.
+        """
+        pass
+
+    @abc.abstractmethod
+    def ioperm(self, port: int, num: int, turn_on: bool) -> None:
+        """
+        Set I/O permissions for a range of ports.
+        :param port: Start of port range.
+        :param num: Number of ports to set permissions for.
+        :param turn_on: Whether to turn on or off permissions.
+        """
+        pass
+
+    @abc.abstractmethod
+    def iopl(self, level: int) -> None:
+        """
+        Set I/O permissions level.
+        :param level: Permissions level.
+        """
+        pass