Optimize rgb565 serialization

library/lcd/lcd_comm_rev_a.py

@@ -16,10 +16,10 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
-import struct
 import time
 
 from serial.tools.list_ports import comports
+import numpy as np
 
 from library.lcd.lcd_comm import *
 from library.log import logger
@@ -130,54 +130,78 @@ def SetOrientation(self, orientation: Orientation = Orientation.PORTRAIT):
         byteBuffer[10] = (height & 255)
         self.lcd_serial.write(bytes(byteBuffer))
 
+    @staticmethod
+    def imageToRGB565LE(image: Image):
+        if image.mode not in ["RGB", "RGBA"]:
+            # we need the first 3 channels to be R, G and B
+            image = image.convert("RGB")
+
+        rgb = np.asarray(image)
+
+        # flatten the first 2 dimensions (width and height) into a single stream
+        # of RGB pixels
+        rgb = rgb.reshape((image.size[1] * image.size[0], -1))
+
+        # extract R, G, B channels and promote them to 16 bits
+        r = rgb[:, 0].astype(np.uint16)
+        g = rgb[:, 1].astype(np.uint16)
+        b = rgb[:, 2].astype(np.uint16)
+
+        # construct RGB565
+        r = (r >> 3)
+        g = (g >> 2)
+        b = (b >> 3)
+        rgb565 = (r << 11) | (g << 5) | b
+
+        # serialize to little-endian
+        return rgb565.newbyteorder('<').tobytes()
+
     def DisplayPILImage(
             self,
             image: Image,
             x: int = 0, y: int = 0,
             image_width: int = 0,
             image_height: int = 0
     ):
+        width, height = self.get_width(), self.get_height()
+
         # If the image height/width isn't provided, use the native image size
         if not image_height:
             image_height = image.size[1]
         if not image_width:
             image_width = image.size[0]
 
-        # If our image is bigger than our display, resize it to fit our screen
-        if image.size[1] > self.get_height():
-            image_height = self.get_height()
-        if image.size[0] > self.get_width():
-            image_width = self.get_width()
-
-        assert x <= self.get_width(), 'Image X coordinate must be <= display width'
-        assert y <= self.get_height(), 'Image Y coordinate must be <= display height'
+        assert x <= width, 'Image X coordinate must be <= display width'
+        assert y <= height, 'Image Y coordinate must be <= display height'
         assert image_height > 0, 'Image height must be > 0'
         assert image_width > 0, 'Image width must be > 0'
 
+        # If our image size + the (x, y) position offsets are bigger than
+        # our display, reduce the image size to fit our screen
+        if x + image_width > width:
+            image_width = width - x
+        if y + image_height > height:
+            image_height = height - y
+
+        if image_width != image.size[0] or image_height != image.size[1]:
+            image = image.crop((0, 0, image_width, image_height))
+
         (x0, y0) = (x, y)
         (x1, y1) = (x + image_width - 1, y + image_height - 1)
 
-        self.SendCommand(Command.DISPLAY_BITMAP, x0, y0, x1, y1)
-
-        pix = image.load()
-        line = bytes()
+        rgb565le = self.imageToRGB565LE(image)
 
         # Lock queue mutex then queue all the requests for the image data
         with self.update_queue_mutex:
-            for h in range(image_height):
-                for w in range(image_width):
-                    R = pix[w, h][0] >> 3
-                    G = pix[w, h][1] >> 2
-                    B = pix[w, h][2] >> 3
-
-                    rgb = (R << 11) | (G << 5) | B
-                    line += struct.pack('<H', rgb)
+            self.SendCommand(Command.DISPLAY_BITMAP, x0, y0, x1, y1)
 
-                    # Send image data by multiple of "display width" bytes
-                    if len(line) >= self.get_width() * 8:
-                        self.SendLine(line)
-                        line = bytes()
+            # Send image data by multiple of "display width" bytes
+            start = 0
+            end = width * 8
+            while end <= len(rgb565le):
+                self.SendLine(rgb565le[start:end])
+                start, end = end, end + width * 8
 
             # Write last line if needed
-            if len(line) > 0:
-                self.SendLine(line)
+            if start != len(rgb565le):
+                self.SendLine(rgb565le[start:])

requirements.txt

@@ -1,6 +1,7 @@
 # Python packages requirements
 Pillow~=9.5.0         # Image generation
 pyserial~=3.5         # Serial linl to communicate with the display
+numpy~=1.19           # Efficient image serialization
 PyYAML~=6.0           # For themes files
 psutil~=5.9.5         # CPU / disk / network metrics
 GPUtil~=1.4.0         # Nvidia GPU

simple-program.py

@@ -22,6 +22,7 @@
 import os
 import signal
 import sys
+import time
 from datetime import datetime
 
 # Import only the modules for LCD communication
@@ -117,7 +118,11 @@ def sighandler(signum, frame):
         background = f"res/backgrounds/{REVISION}/example{size}_landscape.png"
 
     # Display sample picture
+    logger.debug("setting background picture")
+    start = time.perf_counter()
     lcd_comm.DisplayBitmap(background)
+    end = time.perf_counter()
+    logger.debug(f"background picture set (took {end-start:.3f} s)")
 
     # Display sample text
     lcd_comm.DisplayText("Basic text", 50, 100)
@@ -140,6 +145,7 @@ def sighandler(signum, frame):
     # Display the current time and some progress bars as fast as possible
     bar_value = 0
     while not stop:
+        start = time.perf_counter()
         lcd_comm.DisplayText(str(datetime.now().time()), 160, 2,
                              font="roboto/Roboto-Bold.ttf",
                              font_size=20,
@@ -184,6 +190,8 @@ def sighandler(signum, frame):
                                           background_image=background)
 
         bar_value = (bar_value + 2) % 101
+        end = time.perf_counter()
+        logger.debug(f"refresh done (took {end-start:.3f} s)")
 
     # Close serial connection at exit
     lcd_comm.closeSerial()