Add miniscreen app for onboarding (#102)

debian/pt-web-portal.install

@@ -4,3 +4,4 @@ src/notify-on-port-busy.sh                  /usr/lib/pt-web-portal
 src/server/Pipfile*                         /usr/lib/pt-web-portal
 src/server/run.py                           /usr/lib/pt-web-portal
 src/server/backend                          /usr/lib/pt-web-portal
+src/miniscreen                              /usr/lib/pt-web-portal

src/miniscreen/onboarding/app.py

@@ -0,0 +1,76 @@
+import atexit
+from PIL import (
+    Image,
+    ImageDraw,
+)
+from threading import Thread
+from time import sleep
+
+from pitop import Pitop
+from pitopcommon.logger import PTLogger
+
+from .menus import (
+    ApMenuPage,
+    EthernetMenuPage,
+    InfoMenuPage,
+    Menus,
+    UsbMenuPage,
+)
+
+
+class OnboardingApp:
+    def __init__(self):
+        self.miniscreen = Pitop().miniscreen
+        self.miniscreen.up_button.when_pressed = lambda: self.go_to(self.current_page.type.previous())
+        self.miniscreen.down_button.when_pressed = lambda: self.go_to(self.current_page.type.next())
+        self.pages = {
+            Menus.AP: ApMenuPage(self.miniscreen.size, self.miniscreen.mode),
+            Menus.USB: UsbMenuPage(self.miniscreen.size, self.miniscreen.mode),
+            Menus.ETHERNET: EthernetMenuPage(self.miniscreen.size, self.miniscreen.mode),
+            Menus.INFO: InfoMenuPage(self.miniscreen.size, self.miniscreen.mode),
+        }
+
+        self.current_page = self.pages.get(Menus.AP)
+        self.next_page = None
+
+        self.__auto_play_thread = None
+        self.__stop_thread = False
+        atexit.register(self.stop)
+
+    def start(self):
+        self.__auto_play_thread = Thread(target=self.__run_in_background, args=())
+        self.__auto_play_thread.daemon = True
+        self.__auto_play_thread.start()
+
+    def stop(self):
+        self.__stop_thread = True
+        if self.__auto_play_thread and self.__auto_play_thread.is_alive():
+            self.__auto_play_thread.join()
+
+    def go_to(self, page):
+        self.next_page = self.pages.get(page)
+        PTLogger.info(f"Moving to {self.next_page.type.name} page")
+
+    def __run_in_background(self):
+        try:
+            empty_image = Image.new(self.miniscreen.mode, self.miniscreen.size)
+            force_redraw = False
+            while self.__stop_thread is False:
+                image = empty_image.copy()
+                draw = ImageDraw.Draw(image)
+
+                self.current_page.render(draw, redraw=force_redraw)
+                force_redraw = False
+
+                if self.next_page:
+                    self.current_page = self.next_page
+                    self.next_page = None
+                    force_redraw = True
+                    self.current_page.first_draw = True
+
+                self.miniscreen.device.display(image)
+                sleep(self.current_page.interval)
+        except KeyboardInterrupt:
+            pass
+        finally:
+            self.miniscreen.stop_animated_image()

src/miniscreen/onboarding/connection_methods.py

@@ -0,0 +1,128 @@
+from abc import (
+    ABC,
+    abstractmethod,
+)
+from enum import Enum, auto
+from getpass import getuser
+from ipaddress import ip_address
+
+from pitopcommon.sys_info import (
+    get_address_for_ptusb_connected_device,
+    get_ap_mode_status,
+    get_internal_ip,
+)
+from pitopcommon.pt_os import is_pi_using_default_password
+
+
+class ConnectionMethod(Enum):
+    AP = auto()
+    USB = auto()
+    ETHERNET = auto()
+    NONE = auto()
+
+
+class ConnectionMethodBase(ABC):
+    def __init__(
+        self,
+        connection_method,
+        ip="",
+        interface_name="",
+        metadata=dict()
+    ):
+        self.connection_method = connection_method
+        self.ip = ip
+        self.interface_name = interface_name
+        self.metadata = metadata
+
+    @abstractmethod
+    def update(self):
+        raise NotImplementedError
+
+    @abstractmethod
+    def is_connected(self):
+        raise NotImplementedError
+
+    def __eq__(self, other):
+        return isinstance(other, ConnectionMethodBase) \
+            and self.metadata == other.metadata \
+            and self.connection_method == other.connection_method \
+            and self.ip == other.ip \
+            and self.is_connected() == other.is_connected()
+
+
+class UsbConnection(ConnectionMethodBase):
+    def __init__(self):
+        super(UsbConnection, self).__init__(
+            connection_method=ConnectionMethod.USB,
+            ip="",
+            interface_name="ptusb0",
+            metadata={
+                "username": "pi" if getuser() == "root" else getuser(),
+                "password": "pi-top" if is_pi_using_default_password() is True else "********",
+            })
+        self.connected_device_ip = ""
+        self.update()
+
+    def update(self):
+        try:
+            self.ip = ip_address(get_internal_ip(iface=self.interface_name))
+            self.connected_device_ip = get_address_for_ptusb_connected_device()
+        except Exception:
+            self.ip = ""
+            self.connected_device_ip = ""
+
+    def is_connected(self):
+        return self.connected_device_ip != ""
+
+    def __eq__(self, other):
+        return isinstance(other, UsbConnection) \
+            and self.metadata == other.metadata \
+            and self.connection_method == other.connection_method \
+            and self.ip == other.ip \
+            and self.is_connected() == other.is_connected() \
+            and self.connected_device_ip == other.connected_device_ip
+
+
+class ApConnection(ConnectionMethodBase):
+    def __init__(self):
+        super(ApConnection, self).__init__(
+            connection_method=ConnectionMethod.AP,
+            ip="",
+            interface_name="wlan_ap0",
+            metadata=get_ap_mode_status())
+        self.update()
+
+    def update(self):
+        self.metadata = get_ap_mode_status()
+        try:
+            self.ip = ip_address(get_internal_ip(iface=self.interface_name))
+        except Exception:
+            self.ip = ""
+
+    def is_connected(self):
+        try:
+            return get_ap_mode_status().get("state", "") == "active"
+        except Exception:
+            return False
+
+
+class EthernetConnection(ConnectionMethodBase):
+    def __init__(self):
+        super(EthernetConnection, self).__init__(
+            connection_method=ConnectionMethod.ETHERNET,
+            ip="",
+            interface_name="eth0",
+            metadata={
+                "username": "pi" if getuser() == "root" else getuser(),
+                "password": "pi-top" if is_pi_using_default_password() is True else "********",
+            })
+        self.update()
+
+    def update(self):
+        try:
+            self.ip = ip_address(get_internal_ip(iface=self.interface_name))
+        except Exception:
+            self.ip = ""
+
+    def is_connected(self):
+        return self.ip != ""

src/miniscreen/onboarding/helpers.py

@@ -0,0 +1,40 @@
+from PIL import (
+    Image,
+    ImageFont,
+)
+
+
+def draw_text(canvas, text, xy, font_size=12):
+    font = ImageFont.truetype("/usr/share/fonts/truetype/roboto/unhinted/RobotoTTF/Roboto-Regular.ttf", size=font_size)
+    canvas.text(
+        text=str(text),
+        xy=xy,
+        fill=1,
+        font=font,
+        anchor=None,
+        spacing=0,
+        align="left",
+        features=None,
+        font_size=font_size,
+    )
+
+
+def process_image(image_to_process, size, mode):
+    if image_to_process.size == size:
+        image = image_to_process
+        if image.mode != mode:
+            image = image.convert(mode)
+    else:
+        image = Image.new(
+            mode,
+            size,
+            "black"
+        )
+        image.paste(
+            image_to_process.resize(
+                size,
+                resample=Image.NEAREST
+            )
+        )
+
+    return image