Merge branch 'main' into config_app_2

app/layout.css

@@ -0,0 +1,43 @@
+Screen {
+    layout: vertical;
+    background: rgb(32, 0, 20);
+}
+
+Processing {
+    layout: horizontal;
+}
+Ended {
+    layout: horizontal;
+}
+FileSelector {
+    layout: horizontal;
+    height: 3;
+}
+
+LogScreen {
+    background: rgb(32, 0, 20);
+}
+
+.box {
+    height: 1fr;
+}
+
+.box2 {
+    height: 100%;
+    width: 1fr;
+    /* background: rgb(102, 1, 63); */
+    border: solid rgb(255, 0, 106);
+}
+.data_text {
+    height: 100%;
+    width: 1fr;
+    color: rgb(255, 160, 200);
+}
+.fileselectorbutton {
+    height: 3;
+    width: 1fr;
+    border: solid rgb(255, 0, 106);
+    margin-left: 1;
+    margin-right: 1;
+    background: rgb(167, 0, 69);
+}

app/main.py

@@ -0,0 +1,193 @@
+from textual.app import App, ComposeResult
+from textual.containers import ScrollableContainer, Container
+from textual.widgets import Header, Footer, DataTable, Static, Button
+from textual.reactive import reactive
+from textual.message import Message
+from textual.screen import Screen
+from textual.widget import Widget
+from textual import on
+from textual import events
+import systemctl_python
+
+
+class Devices_list(Static):
+    dev_type_g = ""
+    devices=reactive([])
+
+    def __init__(self, dev_type):
+        self.dev_type = dev_type
+        super().__init__()
+    def update_devices(self) -> None:
+        self.devices = systemctl_python.list_working_units("rpi-sb-" + self.dev_type + "*")
+    def watch_devices(self, devices) -> None:
+        """Called when the devices variable changes"""
+        text = ""
+        for i in range(len(devices)):
+            text += devices[i] + "\n"
+        self.styles.height = len(devices)
+        self.update(text)
+
+    def on_mount(self) -> None:
+        self.set_interval(1/20, self.update_devices)
+
+ROWS = [
+    ("Serial Number",),
+]
+
+class CompletedDevicesList(Widget):
+    dev_type_g = ""
+    devices=reactive([])
+    def compose(self) -> ComposeResult:
+         yield DataTable()
+    def __init__(self, dev_type):
+        self.dev_type = dev_type
+        super().__init__()
+    def update_devices(self) -> None:
+        self.devices = systemctl_python.list_completed_devices()
+    def watch_devices(self, devices: list[str]) -> None:
+        """Called when the devices variable changes"""
+        table = self.query_one(DataTable)
+        table_devices = []
+        for device in self.devices:
+            table_devices.append((device, ))
+        table.clear()
+        table.add_rows(table_devices)
+
+    def on_mount(self) -> None:
+        table = self.query_one(DataTable)
+        table.add_columns(*ROWS[0])
+        table.add_rows(ROWS[1:])
+        self.set_interval(1/20, self.update_devices)
+
+class Failed_devices_list(Static):
+    dev_type_g = ""
+    devices=reactive([])
+    def compose(self) -> ComposeResult:
+         yield DataTable()
+    def __init__(self, dev_type):
+        self.dev_type = dev_type
+        super().__init__()
+    def update_devices(self) -> None:
+        self.devices = systemctl_python.list_failed_devices()
+    def watch_devices(self, devices: list[str]) -> None:
+        """Called when the devices variable changes"""
+        table = self.query_one(DataTable)
+        table_devices = []# [("TEST",), ("TEST",)]
+        for device in self.devices:
+            table_devices.append((device, ))
+        table.clear()
+        table.add_rows(table_devices)
+
+    def on_mount(self) -> None:
+        table = self.query_one(DataTable)
+        table.add_columns(*ROWS[0])
+        table.add_rows(ROWS[1:])
+        self.set_interval(1/20, self.update_devices)
+
+class Triage_Box(Static):
+    def compose(self) -> ComposeResult:
+        yield ScrollableContainer(Static("Triaging \n----------------"), Devices_list(dev_type="triage"))
+
+class Keywrite_Box(Static):
+    def compose(self) -> ComposeResult:
+        yield ScrollableContainer(Static("Keywriting \n----------------"), Devices_list(dev_type="keywriter"))
+
+class Provision_Box(Static):
+    def compose(self) -> ComposeResult:
+        yield ScrollableContainer(Static("Provisioning \n----------------"), Devices_list(dev_type="provision"))
+
+
+class Completed_Box(Static):
+    def compose(self) -> ComposeResult:
+        yield ScrollableContainer(Static("Completed \n----------------"), CompletedDevicesList(dev_type="provision"))
+
+class Failed_Box(Static):
+    def compose(self) -> ComposeResult:
+        yield ScrollableContainer(Static("Failed \n----------------"), Failed_devices_list(dev_type="provision"))
+
+class Processing(Static):
+    def compose(self) -> ComposeResult:
+        yield Triage_Box("1", classes="box2")
+        yield Keywrite_Box("2", classes="box2")
+        yield Provision_Box("3", classes="box2")
+
+class Ended(Static):
+    def compose(self) -> ComposeResult:
+        yield Completed_Box("1", classes="box2")
+        yield Failed_Box("2", classes="box2")
+
+class FileSelector(Container):
+    def __init__(self, filelist):
+        self.filelist = filelist
+        self.selected_file = None
+        self.id_to_filename = {}
+        super().__init__()
+    def compose(self) -> ComposeResult:
+        """Create child widgets for the app."""
+        # List files in the directory
+        for file in self.filelist:
+            self.id_to_filename.update([(file.replace(".", ""),  file)])
+            yield Button(file, id=file.replace(".", ""), classes="fileselectorbutton")
+    def get_filename_from_id(self, id) -> str:
+        return self.id_to_filename[id]
+
+class MainScreen(Screen):
+    def compose(self) -> ComposeResult:
+            """Create child widgets for the app."""
+            yield Header()
+            yield Footer()
+            yield Processing("Processing", classes="box")
+            yield Ended("Completed", classes="box")
+    def action_goto_log(self) -> None:
+        self.dismiss(self.query_one(Ended).get_device())
+
+    @on(DataTable.CellSelected)
+    def on_cell_selected(self, event: DataTable.CellSelected) -> None:  
+        self.dismiss(event.value)
+
+class LogScreen(Screen):
+    def __init__(self, device_name):
+        self.device_name = device_name
+        super().__init__()
+
+    def compose(self) -> ComposeResult:
+        """Create child widgets for the app."""
+        yield Header()
+        yield Footer()
+        yield Static("This is the log screen for device: " + self.device_name, id="header_string")
+        yield FileSelector(filelist=systemctl_python.list_device_files(self.device_name))
+        yield ScrollableContainer(Static(" ", id="file_contents"))
+
+    def on_button_pressed(self, event: Button.Pressed) -> None:
+        static = self.query_one("#file_contents")
+        fileselector = self.query_one("FileSelector")
+        # Need to read the file into this container now!
+        contents = systemctl_python.read_device_file(self.device_name, fileselector.get_filename_from_id(event.button.id))
+        static.update(contents)
+
+    def on_screen_resume(self) -> None:  
+        static = self.query_one("#header_string")
+        static.update(self.device_name)
+
+
+class App(App):
+    """A Textual app to manage stopwatches."""
+    CSS_PATH = "layout.css"
+    BINDINGS = [("m", "mainscreen", "Main Screen"), ("q", "quit", "Quit")]
+    SCREENS = {"MainScreen": MainScreen(), "LogScreen": LogScreen("unknown-serial")}    
+
+    def on_mount(self) -> None:
+        self.title = "rpi-sb-provisioner"
+        self.push_screen(LogScreen(device_name="INIT"))
+        self.push_screen(MainScreen(), self.action_logscreen)
+
+    def action_mainscreen(self):
+        self.pop_screen()
+        self.push_screen(MainScreen(), self.action_logscreen)
+
+    def action_logscreen(self, device: str):
+        self.push_screen(LogScreen(device))
+
+if __name__ == "__main__":
+    app = App()
+    app.run()

app/systemctl_python.py

@@ -0,0 +1,88 @@
+import subprocess
+from os import listdir, path
+
+def list_rpi_sb_units(service_name):
+    output = subprocess.run(["systemctl", "list-units", service_name, "-l", "--all", "--no-pager"], capture_output=True)
+    triage=[]
+    keywriter=[]
+    provisioner=[]
+
+    lines = output.stdout.decode().split("\n")
+    for line in lines:
+        if "rpi-sb-" in line:
+            name=line[line.find("rpi-sb-"):line.find(".service")]
+            if "triage" in name:
+                triage.append(name.replace("rpi-sb-triage@", ""))
+            if "keywriter" in name:
+                keywriter.append(name.replace("rpi-sb-keywriter@", ""))
+            if "provisioner" in name:
+                provisioner.append(name.replace("rpi-sb-provisioner@", ""))
+    return [triage, keywriter, provisioner]
+
+def list_working_units(service_name):
+    output = subprocess.run(["systemctl", "list-units", service_name, "-l", "--all", "--no-pager"], capture_output=True)
+    units=[]
+    lines = output.stdout.decode().split("\n")
+    for line in lines:
+        if "rpi-sb-" in line:
+            if not("failed" in line):
+                name=line[line.find("rpi-sb-"):line.find(".service")]
+                units.append(name)
+    return units
+
+def list_failed_units(service_name):
+    output = subprocess.run(["systemctl", "list-units", service_name, "-l", "--all", "--no-pager"], capture_output=True)
+    units=[]
+    lines = output.stdout.decode().split("\n")
+    for line in lines:
+        if "rpi-sb-" in line:
+            if "failed" in line:
+                name=line[line.find("rpi-sb-"):line.find(".service")]
+                units.append(name)
+    return units
+
+def list_seen_devices():
+    if path.exists("/var/log/rpi-sb-provisioner/"):
+        devices = listdir("/var/log/rpi-sb-provisioner")
+        return devices
+    else:
+        return []
+
+def list_completed_devices():
+    all_devices = list_seen_devices()
+    completed_devices = []
+    for device in all_devices:
+        if path.exists("/var/log/rpi-sb-provisioner/" + device + "/success"):
+            f = open("/var/log/rpi-sb-provisioner/" + device + "/success", "r")
+            status = f.read()
+            if "1" in status:
+                completed_devices.append(device)
+            f.close()
+    return completed_devices
+
+def list_failed_devices():
+    all_devices = list_seen_devices()
+    failed_devices = []
+    for device in all_devices:
+        if path.exists("/var/log/rpi-sb-provisioner/" + device + "/finished"):
+            if not(path.exists("/var/log/rpi-sb-provisioner/" + device + "/success")):
+                f = open("/var/log/rpi-sb-provisioner/" + device + "/finished", "r")
+                status = f.read()
+                if "1" in status:
+                    failed_devices.append(device)
+                f.close()
+    return failed_devices
+
+def list_device_files(device_name):
+    if path.exists("/var/log/rpi-sb-provisioner/" + device_name):
+        return listdir("/var/log/rpi-sb-provisioner/" + device_name)
+    else:
+        return []
+
+def read_device_file(device_name, filename):
+    contents = "Unable to read/open file!"
+    if path.exists("/var/log/rpi-sb-provisioner/" + device_name + "/" + filename):
+        f = open("/var/log/rpi-sb-provisioner/" + device_name + "/" + filename, "r")
+        contents = f.read()
+        f.close()
+    return contents