Phui changes

docker-compose.yml

@@ -1,7 +1,7 @@
 services:
   yarilo:
     image: typicalam/yarilo:latest
-    command: --battery_file=/app/battery/battery_level --oid_file=/app/data/oid.txt --save_path=/app/saves --db_file=/app/saves/yarilo_database.db --sniff_file=/tmp/pcap/wireshark_sample.pcap
+    command: --log_level=debug --battery_file=/app/battery/battery_level --oid_file=/app/data/oid.txt --save_path=/app/saves --db_file=/app/saves/yarilo_database.db --iface=wlan1
     volumes:
       - /tmp/saves:/app/saves
       - ./pcap:/tmp/pcap

gui-2.0/.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+.lgd-nfy0

gui-2.0/button_handler.py

@@ -0,0 +1,29 @@
+import RPi.GPIO as GPIO
+
+BUTTON_PINS = {
+    17: "ACCEPT",
+    22: "REFUSE",
+    26: "DOWN",
+    13: "LEFT",
+    6: "RIGHT",
+    5: "UP"
+}
+
+class ButtonHandler:
+    def __init__(self, callback):
+        """
+        :param callback: A function that accepts (channel, button_name)
+        """
+        self.callback = callback
+        GPIO.setmode(GPIO.BCM)
+        for pin in BUTTON_PINS:
+            GPIO.setup(pin, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
+            try:
+                GPIO.remove_event_detect(pin)
+            except RuntimeError:
+                pass
+            GPIO.add_event_detect(pin, GPIO.RISING, callback=self.handle_button, bouncetime=100)
+
+    def handle_button(self, channel):
+        button_name = BUTTON_PINS.get(channel, "UNKNOWN")
+        self.callback(channel, button_name)

gui-2.0/client.py

@@ -0,0 +1,141 @@
+import grpc
+import os
+import sys
+sys.path.append(os.path.dirname(__file__))
+import service_pb2, service_pb2_grpc
+
+class Client:
+    def __init__(self, host='localhost', port=9090):
+        self.channel = grpc.insecure_channel(f'{host}:{port}')
+        self.stub = service_pb2_grpc.SnifferStub(self.channel)
+        # Cache the primary sniffer if available
+        self._sniffer_uuid = None
+
+    def _get_primary_sniffer_uuid(self):
+        if self._sniffer_uuid is None:
+            response = self.stub.SnifferList(service_pb2.Empty())
+            if response.sniffers:
+                self._sniffer_uuid = response.sniffers[0].uuid
+            else:
+                raise ValueError("No sniffers found.")
+        return self._sniffer_uuid
+
+    def is_connected(self):
+        try:
+            self.stub.SnifferList(service_pb2.Empty())
+            return True
+        except grpc.RpcError as e:
+            if e.code() == grpc.StatusCode.UNAVAILABLE:
+                print("gRPC Error: Unable to connect to the server.")
+            else:
+                print(f"gRPC Error: {e.details()} (Code: {e.code()})")
+            return False
+
+    def get_sniffer_list(self):
+        response = self.stub.SnifferList(service_pb2.Empty())
+        if not response.sniffers:
+            return "No sniffers found."
+        sniffers = [
+            f"uuid: {sniffer.uuid} interface name: {sniffer.net_iface_name}\nfilename: {sniffer.filename}\n"
+            for sniffer in response.sniffers
+        ]
+        return "Sniffers:\n" + "\n".join(sniffers)
+
+    def get_access_point_list(self):
+        try:
+            uuid = self._get_primary_sniffer_uuid()
+            request = service_pb2.APGetRequest(sniffer_uuid=uuid)
+            response = self.stub.AccessPointList(request)
+            if not response.nets:
+                return "No access points found."
+            access_points = [f"{ap.ssid} - {ap.bssid}\n" for ap in response.nets]
+            return "Access Points:\n" + "".join(access_points)
+        except ValueError as e:
+            return str(e)
+
+    def get_networks_list(self):
+        try:
+            uuid = self._get_primary_sniffer_uuid()
+            request = service_pb2.APGetRequest(sniffer_uuid=uuid)
+            response = self.stub.AccessPointList(request)
+            if not response.nets:
+                return "No networks found."
+            networks = [str([network.ssid, network.bssid]) for network in response.nets]
+            return networks
+        except ValueError as e:
+            return str(e)
+
+    def create_recording(self):
+        try:
+            uuid = self._get_primary_sniffer_uuid()
+            request = service_pb2.RecordingCreateRequest(
+                sniffer_uuid=uuid,
+                name='Manual_recording',
+                raw=True
+            )
+            response = self.stub.RecordingCreate(request)
+            return str(response)
+        except ValueError as e:
+            return str(e)
+
+    def create_APrecording(self, bssid):
+        try:
+            uuid = self._get_primary_sniffer_uuid()
+            request = service_pb2.APCreateRecordingRequest(
+                sniffer_uuid=uuid,
+                name='AccessPoint_recording',
+                bssid=bssid,
+                raw=True
+            )
+            response = self.stub.RecordingCreate(request)
+            return str(response)
+        except ValueError as e:
+            return str(e)
+
+    def get_battery(self):
+        try:
+            response = self.stub.BatteryGetLevel(service_pb2.Empty())
+            # Use the float value directly from response.percentage
+            formated_resp = f"{response.percentage:.0f}"
+            formated_resp = int(formated_resp)
+            return formated_resp
+        except grpc.RpcError as e:
+            return "Get battery error"
+
+    def start_focus(self, network):
+        try:
+            uuid = self._get_primary_sniffer_uuid()
+            request = service_pb2.FocusStartRequest(
+                sniffer_uuid=uuid,
+                bssid=network
+            )
+            response = self.stub.FocusStart(request)
+            print(response)
+            return str(response)
+        except grpc.RpcError as e:
+            print(e)
+            return "Start focus error"
+
+    def get_active_focus(self):
+        try:
+            uuid = self._get_primary_sniffer_uuid()
+            response = self.stub.FocusGetActive(uuid)
+            return str(response)
+        except grpc.RpcError as e:
+            print(e)
+            return "Get active focus error"
+
+    def ignore_AP(self, network):
+        try:
+            uuid = self._get_primary_sniffer_uuid()
+            request = service_pb2.APIgnoreRequest(
+                sniffer_uuid=uuid,
+                use_ssid=False,
+                bssid=network,
+                ssid=""
+            )
+            response = self.stub.AccessPointIgnore(request)
+            return str(response)
+        except grpc.RpcError as e:
+            print(e)
+            return "Ignore AP error"

gui-2.0/display.py

@@ -0,0 +1,57 @@
+from lib import LCD_2inch4
+import spidev
+from PIL import Image, ImageDraw, ImageFont
+
+# Constants
+RST = 27
+DC = 25
+BL = 18
+BUS = 0
+DEVICE = 0
+FONT_SIZE = 25
+
+class Display:
+    def __init__(self):
+        self.disp = LCD_2inch4.LCD_2inch4(
+            spi=spidev.SpiDev(BUS, DEVICE), spi_freq=40000000, rst=RST, dc=DC, bl=BL)
+        self.disp.Init()
+        self.disp.clear()
+
+        self.elements = []
+
+        # Set up drawing
+        self.width = 320
+        self.height = 240
+        self.image = Image.new("RGB", (self.width, self.height), "white")
+        self.draw = ImageDraw.Draw(self.image)
+
+        # Load font
+        self.font = ImageFont.load_default()
+
+        # Note: GPIO/button handling is now managed via ButtonHandler
+
+    def update(self):
+        image_to_disp = self.image.rotate(180)
+        self.disp.ShowImage(image_to_disp)
+
+    def clear(self):
+        self.draw.rectangle((0, 0, self.width, self.height), fill="white")
+
+    def draw_text(self, text, position, color="white"):
+        self.draw.text(position, text, fill=color, font=self.font)
+
+    def draw_rectangle(self, bbox, color="white", outline=None):
+        self.draw.rectangle(bbox, fill=color, outline=outline)
+
+    def add_element(self, element):
+        self.elements.append(element)
+
+    def render(self):
+        self.clear()
+        for element in self.elements:
+            element.draw(self.draw)
+        self.update()
+
+    def cleanup(self):
+        self.disp.reset()
+        self.disp.clear()