Merge pull request #641 from hchargois/numpy-everywhere

Use Numpy for all revisions

Author: mathoudebine

library/lcd/lcd_comm_rev_a.py

@@ -21,9 +21,9 @@
 from typing import Optional
 
 from serial.tools.list_ports import comports
-import numpy as np
 
 from library.lcd.lcd_comm import *
+from library.lcd.serialize import image_to_RGB565, chunked
 from library.log import logger
 
 
@@ -173,32 +173,6 @@ def SetOrientation(self, orientation: Orientation = Orientation.PORTRAIT):
         byteBuffer[10] = (height & 255)
         self.serial_write(bytes(byteBuffer))
 
-    @staticmethod
-    def imageToRGB565LE(image: 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.astype('<u2').tobytes()
-
     def DisplayPILImage(
             self,
             image: Image.Image,
@@ -232,20 +206,12 @@ def DisplayPILImage(
         (x0, y0) = (x, y)
         (x1, y1) = (x + image_width - 1, y + image_height - 1)
 
-        rgb565le = self.imageToRGB565LE(image)
+        rgb565le = image_to_RGB565(image, "little")
 
         self.SendCommand(Command.DISPLAY_BITMAP, x0, y0, x1, y1)
 
         # Lock queue mutex then queue all the requests for the image data
         with self.update_queue_mutex:
-
             # 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 start != len(rgb565le):
-                self.SendLine(rgb565le[start:])
+            for chunk in chunked(rgb565le, width * 8):
+                self.SendLine(chunk)

library/lcd/lcd_comm_rev_b.py

@@ -17,11 +17,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
-
 from serial.tools.list_ports import comports
 
 from library.lcd.lcd_comm import *
+from library.lcd.serialize import image_to_RGB565, chunked
 from library.log import logger
 
 
@@ -194,6 +193,13 @@ def SetOrientation(self, orientation: Orientation = Orientation.PORTRAIT):
         else:
             self.SendCommand(Command.SET_ORIENTATION, payload=[OrientationValueRevB.ORIENTATION_LANDSCAPE])
 
+    def serialize_image(self, image: Image.Image, height: int, width: int) -> bytes:
+        if image.width != width or image.height != height:
+            image = image.crop((0, 0, width, height))
+        if self.orientation == Orientation.REVERSE_PORTRAIT or self.orientation == Orientation.REVERSE_LANDSCAPE:
+            image = image.rotate(180)
+        return image_to_RGB565(image, "big")
+
     def DisplayPILImage(
             self,
             image: Image.Image,
@@ -231,34 +237,11 @@ def DisplayPILImage(
                                   (y0 >> 8) & 255, y0 & 255,
                                   (x1 >> 8) & 255, x1 & 255,
                                   (y1 >> 8) & 255, y1 & 255])
-        pix = image.load()
-        line = bytes()
+
+        rgb565be = self.serialize_image(image, image_height, image_width)
 
         # 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):
-                    if self.orientation == Orientation.PORTRAIT or self.orientation == Orientation.LANDSCAPE:
-                        R = pix[w, h][0] >> 3
-                        G = pix[w, h][1] >> 2
-                        B = pix[w, h][2] >> 3
-                    else:
-                        # Manage reverse orientations from software, because display does not manage it
-                        R = pix[image_width - w - 1, image_height - h - 1][0] >> 3
-                        G = pix[image_width - w - 1, image_height - h - 1][1] >> 2
-                        B = pix[image_width - w - 1, image_height - h - 1][2] >> 3
-
-                    # Color information is 0bRRRRRGGGGGGBBBBB
-                    # Revision A: Encode in Little-Endian (native x86/ARM encoding)
-                    # Revition B: Encode in Big-Endian
-                    rgb = (R << 11) | (G << 5) | B
-                    line += struct.pack('>H', rgb)
-
-                    # Send image data by multiple of "display width" bytes
-                    if len(line) >= self.get_width() * 8:
-                        self.SendLine(line)
-                        line = bytes()
-
-            # Write last line if needed
-            if len(line) > 0:
-                self.SendLine(line)
+            # Send image data by multiple of "display width" bytes
+            for chunk in chunked(rgb565be, self.get_width() * 8):
+                self.SendLine(chunk)

library/lcd/lcd_comm_rev_c.py

@@ -22,13 +22,14 @@
 import time
 from enum import Enum
 from math import ceil
-from typing import Optional
+from typing import Optional, Tuple
 
 import serial
 from PIL import Image
 from serial.tools.list_ports import comports
 
 from library.lcd.lcd_comm import Orientation, LcdComm
+from library.lcd.serialize import image_to_BGRA, image_to_BGR, chunked
 from library.log import logger
 
 
@@ -282,6 +283,9 @@ def DisplayPILImage(
         if image.size[0] > self.get_width():
             image_width = self.get_width()
 
+        if image_width != image.size[0] or image_height != image.size[1]:
+            image = image.crop((0, 0, image_width, image_height))
+
         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 image_height > 0, 'Image height must be > 0'
@@ -293,76 +297,67 @@ def DisplayPILImage(
                 self._send_command(Command.START_DISPLAY_BITMAP, padding=Padding.START_DISPLAY_BITMAP)
                 self._send_command(Command.DISPLAY_BITMAP)
                 self._send_command(Command.SEND_PAYLOAD,
-                                   payload=bytearray(self._generate_full_image(image, self.orientation)),
+                                   payload=bytearray(self._generate_full_image(image)),
                                    readsize=1024)
                 self._send_command(Command.QUERY_STATUS, readsize=1024)
         else:
             with self.update_queue_mutex:
-                img, pyd = self._generate_update_image(image, x, y, Count.Start, Command.UPDATE_BITMAP,
-                                                       self.orientation)
+                img, pyd = self._generate_update_image(image, x, y, Count.Start, Command.UPDATE_BITMAP)
                 self._send_command(Command.SEND_PAYLOAD, payload=pyd)
                 self._send_command(Command.SEND_PAYLOAD, payload=img)
                 self._send_command(Command.QUERY_STATUS, readsize=1024)
             Count.Start += 1
 
-    @staticmethod
-    def _generate_full_image(image: Image.Image, orientation: Orientation = Orientation.PORTRAIT):
-        if orientation == Orientation.PORTRAIT:
+    def _generate_full_image(self, image: Image.Image) -> bytes:
+        if self.orientation == Orientation.PORTRAIT:
             image = image.rotate(90, expand=True)
-        elif orientation == Orientation.REVERSE_PORTRAIT:
+        elif self.orientation == Orientation.REVERSE_PORTRAIT:
             image = image.rotate(270, expand=True)
-        elif orientation == Orientation.REVERSE_LANDSCAPE:
+        elif self.orientation == Orientation.REVERSE_LANDSCAPE:
             image = image.rotate(180)
 
-        image_data = image.convert("RGBA").load()
-        image_ret = ''
-        for y in range(image.height):
-            for x in range(image.width):
-                pixel = image_data[x, y]
-                image_ret += f'{pixel[2]:02x}{pixel[1]:02x}{pixel[0]:02x}{pixel[3]:02x}'
+        bgra_data = image_to_BGRA(image)
 
-        hex_data = bytearray.fromhex(image_ret)
-        return b'\x00'.join(hex_data[i:i + 249] for i in range(0, len(hex_data), 249))
+        return b'\x00'.join(chunked(bgra_data, 249))
 
-    def _generate_update_image(self, image, x, y, count, cmd: Optional[Command] = None,
-                               orientation: Orientation = Orientation.PORTRAIT):
+    def _generate_update_image(
+        self, image: Image.Image, x: int, y: int, count: int, cmd: Optional[Command] = None
+    ) -> Tuple[bytearray, bytearray]:
         x0, y0 = x, y
 
-        if orientation == Orientation.PORTRAIT:
+        if self.orientation == Orientation.PORTRAIT:
             image = image.rotate(90, expand=True)
             x0 = self.get_width() - x - image.height
-        elif orientation == Orientation.REVERSE_PORTRAIT:
+        elif self.orientation == Orientation.REVERSE_PORTRAIT:
             image = image.rotate(270, expand=True)
             y0 = self.get_height() - y - image.width
-        elif orientation == Orientation.REVERSE_LANDSCAPE:
+        elif self.orientation == Orientation.REVERSE_LANDSCAPE:
             image = image.rotate(180, expand=True)
             y0 = self.get_width() - x - image.width
             x0 = self.get_height() - y - image.height
-        elif orientation == Orientation.LANDSCAPE:
+        elif self.orientation == Orientation.LANDSCAPE:
             x0, y0 = y, x
 
-        img_raw_data = []
-        image_data = image.convert("RGBA").load()
-        for h in range(image.height):
-            img_raw_data.append(f'{((x0 + h) * self.display_height) + y0:06x}{image.width:04x}')
-            for w in range(image.width):
-                current_pixel = image_data[w, h]
-                img_raw_data.append(f'{current_pixel[2]:02x}{current_pixel[1]:02x}{current_pixel[0]:02x}')
+        img_raw_data = bytearray()
+        bgr_data = image_to_BGR(image)
+        for h, line in enumerate(chunked(bgr_data, image.width * 3)):
+            img_raw_data += int(((x0 + h) * self.display_height) + y0).to_bytes(3, "big")
+            img_raw_data += int(image.width).to_bytes(2, "big")
+            img_raw_data += line
 
-        image_msg = ''.join(img_raw_data)
-        image_size = f'{int((len(image_msg) / 2) + 2):06x}'  # The +2 is for the "ef69" that will be added later.
+        image_size = int(len(img_raw_data) + 2).to_bytes(3, "big") # The +2 is for the "ef69" that will be added later.
 
         # logger.debug("Render Count: {}".format(count))
         payload = bytearray()
 
         if cmd:
             payload.extend(cmd.value)
-        payload.extend(bytearray.fromhex(image_size))
+        payload.extend(image_size)
         payload.extend(Padding.NULL.value * 3)
         payload.extend(count.to_bytes(4, 'big'))
 
-        if len(image_msg) > 500:
-            image_msg = '00'.join(image_msg[i:i + 498] for i in range(0, len(image_msg), 498))
-        image_msg += 'ef69'
+        if len(img_raw_data) > 250:
+            img_raw_data = bytearray(b'\x00').join(chunked(bytes(img_raw_data), 249))
+        img_raw_data += b'\xef\x69'
 
-        return bytearray.fromhex(image_msg), payload
+        return img_raw_data, payload

library/lcd/lcd_comm_rev_d.py

@@ -16,12 +16,12 @@
 # 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
 from enum import Enum
 
 from serial.tools.list_ports import comports
 
 from library.lcd.lcd_comm import *
+from library.lcd.serialize import image_to_RGB565, chunked
 from library.log import logger
 
 
@@ -109,7 +109,7 @@ def SetBrightness(self, level: int = 25):
         # Convert our brightness % to an absolute value.
         converted_level = level * 5
 
-        level_bytes = bytearray(converted_level.to_bytes(2))
+        level_bytes = bytearray(converted_level.to_bytes(2, "big"))
 
         # Send the command twice because sometimes it is not applied...
         self.SendCommand(cmd=Command.SETBL, payload=level_bytes)
@@ -166,40 +166,22 @@ def DisplayPILImage(
             image_width, image_height = image_height, image_width
 
         # Send bitmap size
-        image_data = bytearray(x0.to_bytes(2))
-        image_data += bytearray(x1.to_bytes(2))
-        image_data += bytearray(y0.to_bytes(2))
-        image_data += bytearray(y1.to_bytes(2))
+        image_data = bytearray()
+        image_data += x0.to_bytes(2, "big")
+        image_data += x1.to_bytes(2, "big")
+        image_data += y0.to_bytes(2, "big")
+        image_data += y1.to_bytes(2, "big")
         self.SendCommand(cmd=Command.BLOCKWRITE, payload=image_data)
 
         # Prepare bitmap data transmission
         self.SendCommand(Command.INTOPICMODE)
 
-        pix = image.load()
-        line = bytes([80])
+        rgb565be = image_to_RGB565(image, "big")
 
         # 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
-
-                    # Color information is 0bRRRRRGGGGGGBBBBB
-                    # Revision A: Encode in Little-Endian (native x86/ARM encoding)
-                    # Revition B: Encode in Big-Endian
-                    rgb = (R << 11) | (G << 5) | B
-                    line += struct.pack('>H', rgb)
-
-                    # Send image data by multiple of 64 bytes + 1 command byte
-                    if len(line) >= 65:
-                        self.SendLine(line[0:64])
-                        line = bytes([80]) + line[64:]
-
-            # Write last line if needed
-            if len(line) > 0:
-                self.SendLine(line)
+            for chunk in chunked(rgb565be, 63):
+                self.SendLine(b"\x50" + chunk)
 
         # Indicate the complete bitmap has been transmitted
         self.SendCommand(Command.OUTPICMODE)

library/lcd/serialize.py

@@ -0,0 +1,58 @@
+from typing import Iterator, Literal
+
+import numpy as np
+from PIL import Image
+
+
+def chunked(data: bytes, chunk_size: int) -> Iterator[bytes]:
+    for i in range(0, len(data), chunk_size):
+        yield data[i : i + chunk_size]
+
+
+def image_to_RGB565(image: Image.Image, endianness: Literal["big", "little"]) -> bytes:
+    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 the correct endianness
+    if endianness == "big":
+        typ = ">u2"
+    else:
+        typ = "<u2"
+    return rgb565.astype(typ).tobytes()
+
+
+def image_to_BGR(image: Image.Image) -> bytes:
+    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)
+    # same as rgb[:, :, [2, 1, 0]] but faster
+    bgr = np.take(rgb, (2, 1, 0), axis=-1)
+    return bgr.tobytes()
+
+
+def image_to_BGRA(image: Image.Image) -> bytes:
+    if image.mode != "RGBA":
+        image = image.convert("RGBA")
+    rgba = np.asarray(image)
+    # same as rgba[:, :, [2, 1, 0, 3]] but faster
+    bgra = np.take(rgba, (2, 1, 0, 3), axis=-1)
+    return bgra.tobytes()