#!/usr/bin/env python # -*- coding: utf-8 -*- import threading import time import RPi.GPIO as GPIO import LCD_1in44 import LCD_Config import cv2 import json import os import shutil import subprocess import picamera from PIL import Image, ImageDraw from scapy.all import RadioTap, Dot11, Dot11Deauth, sendp class RaZero: # Der Konstruktor der Klasse, in dem die Initialisierung der GPIO-Pins, des Displays, der Kamera und anderer Variablen durchgeführt wird. def __init__(self): self.load_configuration("config.json") # Konfiguration laden # Initialisierung des Displays Lcd_ScanDir = LCD_1in44.SCAN_DIR_DFT self.LCD = LCD_1in44.LCD() self.LCD.LCD_Init(Lcd_ScanDir) # Kamera initialisieren self.camera = picamera.PiCamera() self.camera.resolution = self.video_resolution # Kameraauflösung self.display_width = 128 self.display_height = 128 self.init_gpio() # GPIO initialisieren # Eingabegerät über evdev initialisieren #self.evdev_device = evdev.InputDevice('/dev/input/eventX') # Ersetzen Sie 'eventX' durch den tatsächlichen Gerätepfad # Initialisierung von Variablen self.recording = False self.display_on = True self.menu_active = False self.current_menu = [] self.menu_index = 0 # Menüstruktur aus JSON-Datei laden self.load_menu_from_json("menu_structure.json") # Lock für Thread-Synchronisation self.lock = threading.Lock() self.record_interval = 300 # Aufnahmeintervall in Sekunden self.last_record_time = time.time() self.filters = { "none": None, "black_and_white": "none", "polaroid": "polaroid", "infrared": "negative", "blacklight": "negative", "oversaturated": "colorbalance:s=2", "overexposed": "brightness:120", "underexposed": "brightness:80", } self.active_filter = "none" # Aktuell ausgewählter Filter self.init_gpio() # GPIO initialisieren def load_configuration(self, config_file): """ Lädt die Konfiguration aus der JSON-Datei. """ with open(config_file, "r") as f: config = json.load(f) self.nas_address = config["nas_address"] self.local_mount_path = config["local_mount_path"] self.record_interval = config["record_interval"] self.menu_file = config["menu_file"] self.delete_after_upload = config["delete_after_upload"] self.video_resolution = tuple(config["video_resolution"]) self.video_fps = config["video_fps"] self.pins = config["pins"] def init_gpio(self): """ Initialisiert die GPIO-Pins für die Buttons. """ GPIO.setmode(GPIO.BCM) for pin_name, pin_number in self.pins.items(): GPIO.setup(pin_number, GPIO.IN, pull_up_down=GPIO.PUD_UP) def load_menu_from_json(self, filename): """ Lädt die Menüstruktur aus einer JSON-Datei. :param filename: Der Name der JSON-Datei. """ with open(filename, "r") as file: self.menu_structure = json.load(file) # Menüstruktur aus JSON laden def display_menu(self): """ Zeigt das Menü auf dem Display an, einschließlich Symbolen und Untermenüs. """ if not self.menu_active: return self.LCD.LCD_Clear() # Display löschen menu_item = self.current_menu[self.menu_index] # Aktuellen Menüpunkt abrufen self.LCD.LCD_ShowString(menu_item["DisplayName"], 10, 10) # Menüpunkt-Text anzeigen if self.recording: # Zeichnen Sie einen roten Kreis, wenn die Aufnahme läuft self.draw_recording_indicator() # Weitere Anzeigen für Symbole und Untermenüs hier hinzufügen def draw_recording_indicator(self): """ Zeichnet einen roten Kreis auf dem Display als Indikator für laufende Aufnahmen. """ image = Image.new("RGB", (self.LCD.width, self.LCD.height)) # Leeres Bild erstellen draw = ImageDraw.Draw(image) # Zeichnen-Objekt erstellen # Koordinaten für das Zeichnen des roten Kreises (untere rechte Ecke) circle_x = self.LCD.width - 12 circle_y = self.LCD.height - 12 circle_radius = 6 # Kreis zeichnen draw.ellipse((circle_x - circle_radius, circle_y - circle_radius, circle_x + circle_radius, circle_y + circle_radius), fill="red") self.LCD.LCD_ShowImage(image) # Bild auf dem Display anzeigen def update_menu_index(self, direction): """ Aktualisiert den Index des aktuellen Menüpunkts basierend auf der Richtung der Tastenbetätigung. :param direction: Die Richtung der Tastenbetätigung ('up' oder 'down'). """ if direction == "up": self.menu_index = (self.menu_index - 1) % len(self.current_menu) # Index verringern (zirkulär) elif direction == "down": self.menu_index = (self.menu_index + 1) % len(self.current_menu) # Index erhöhen (zirkulär) def handle_menu_select(self): """ Verarbeitet die Auswahl eines Menüpunkts, entweder durch Anzeigen von Untermenüs oder Durchführen von Aktionen. """ if not self.menu_active: # Menü ist nicht aktiv, schalte es ein self.menu_active = True self.current_menu = self.menu_structure # Setze das aktuelle Menü auf das Hauptmenü else: # Menü ist aktiv, verarbeite Auswahl menu_item = self.current_menu[self.menu_index] if "Submenu" in menu_item: # Wenn der Menüpunkt ein Untermenü hat self.current_menu = menu_item["Submenu"] # Wechsle zum Untermenü self.menu_index = 0 # Setze den Index auf den ersten Menüpunkt im Untermenü elif "Action" in menu_item: # Wenn der Menüpunkt eine Aktion hat self.perform_menu_action(menu_item["Action"]) # Führe die Aktion aus def handle_menu_navigation(self, direction): """ Verarbeitet die Navigation im Menü, einschließlich des Zurückkehrens zu übergeordneten Menüpunkten. :param direction: Die Richtung der Tastenbetätigung ('left' oder 'right'). """ if direction == "left": if self.menu_active: self.menu_active = False # Menü deaktivieren, um zur Hauptanzeige zurückzukehren self.current_menu = [] # Setze das aktuelle Menü zurück else: self.display_on = True # Display einschalten, wenn es ausgeschaltet war elif direction == "right": menu_item = self.current_menu[self.menu_index] if "Submenu" in menu_item: # Wenn der Menüpunkt ein Untermenü hat self.current_menu = menu_item["Submenu"] # Wechsle zum Untermenü self.menu_index = 0 # Setze den Index auf den ersten Menüpunkt im Untermenü def perform_menu_action(self, action): """ Führt die Aktion eines ausgewählten Menüpunkts aus. :param action: Die Aktion, die ausgeführt werden soll. """ if action == "start_recording": # Beispiel: Aktion zum Starten der Aufnahme if not self.recording: # Wenn die Aufnahme nicht läuft self.recording = True # Aufnahme starten elif action == "stop_recording": # Beispiel: Aktion zum Beenden der Aufnahme if self.recording: # Wenn die Aufnahme läuft self.recording = False # Aufnahme beenden elif action == "turn_off_display": # Beispiel: Aktion zum Ausschalten des Displays self.display_on = False # Display ausschalten # Fügen Sie hier weitere Aktionen hinzu, die Sie definiert haben def record_thread(self): """ Überwacht die Tasten und startet oder beendet die Aufnahme basierend auf den Benutzerinteraktionen. Führt die Aufnahme in Intervallen von self.record_interval aus. """ while True: with self.lock: current_time = time.time() if self.recording: if current_time - self.last_record_time >= self.record_interval: self.camera.stop_recording() # Aktuelle Aufnahme beenden timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S") filename = f"video_{timestamp}.mp4" source_path = filename self.set_exif_data() # Exif-Daten setzen self.apply_filter(self.active_filter) # Filter anwenden self.camera.start_recording(source_path, format="h264", quality=23, resolution=self.video_resolution, framerate=self.video_fps) self.last_record_time = current_time else: self.camera.stop_recording() # Aufnahme beenden time.sleep(0.1) # Kurze Verzögerung, um die CPU-Belastung zu reduzieren def display_thread(self): """ Überwacht den Tastendruck des Display-Buttons und schaltet das Display ein oder aus. """ while True: with self.lock: if not self.display_on: # Wenn das Display ausgeschaltet ist self.LCD.LCD_Clear() # Display löschen time.sleep(0.1) # Kurze Verzögerung, um die CPU-Belastung zu reduzieren # Überprüfen Sie den Tastendruck auf den Display-Button if GPIO.input(self.KEY3_PIN) == GPIO.LOW: self.display_on = not self.display_on # Schalte den Display-Zustand um time.sleep(0.2) # Vermeiden Sie mehrfache Tastendrücke in schneller Folge def menu_thread(self): """ Überwacht die Tasten für die Menünavigation und -auswahl, zeigt das Menü an und verarbeitet Benutzeraktionen. """ while True: with self.lock: if self.menu_active: # Wenn das Menü aktiv ist self.display_menu() # Menü anzeigen # Überprüfen Sie den Tastendruck auf die Menünavigation und -auswahl if GPIO.input(self.KEY_UP_PIN) == GPIO.LOW: self.update_menu_index("up") # Menüpunkt nach oben wechseln time.sleep(0.2) elif GPIO.input(self.KEY_DOWN_PIN) == GPIO.LOW: self.update_menu_index("down") # Menüpunkt nach unten wechseln time.sleep(0.2) elif GPIO.input(self.KEY_LEFT_PIN) == GPIO.LOW: self.handle_menu_navigation("left") # Zurück oder Display einschalten time.sleep(0.2) elif GPIO.input(self.KEY_RIGHT_PIN) == GPIO.LOW: self.handle_menu_navigation("right") # Untermenü anzeigen time.sleep(0.2) elif GPIO.input(self.KEY_PRESS_PIN) == GPIO.LOW: self.handle_menu_select() # Menüpunkt auswählen time.sleep(0.2) else: self.display_menu() # Menü anzeigen # Überprüfen Sie den Tastendruck auf die Menünavigation und -auswahl if GPIO.input(self.KEY3_PIN) == GPIO.LOW: self.menu_active = True # Menü aktivieren time.sleep(0.2) def capture_scaled_image(self): """ Erfasst das Kamerabild, skaliert es auf die Displaygröße und gibt das Bild zurück. :return: Das skalierte Kamerabild als PIL-Bildobjekt. """ stream = io.BytesIO() # Erstellen Sie einen BytesIO-Stream zum Speichern des Bilds self.camera.capture(stream, format='jpeg', use_video_port=True) # Kamerabild aufnehmen # BytesIO-Stream zurück auf den Anfang setzen und PIL-Bildobjekt erstellen stream.seek(0) image = Image.open(stream) # Skalieren Sie das Bild auf die Displaygröße image = image.resize((self.display_width, self.display_height), Image.ANTIALIAS) return image def set_exif_data(self): """ Setzt die Exif-Daten für das aufgenommene Bild basierend auf den Einstellungen. """ exif_data = { # Fügen Sie hier Ihre gewünschten Exif-Daten hinzu TAGS['Artist']: 'Ihr Name', TAGS['Copyright']: 'Ihr Copyright-Vermerk', TAGS['DateTime']: datetime.datetime.now().strftime('%Y:%m:%d %H:%M:%S'), # Weitere Exif-Tags können hier hinzugefügt werden } self.camera._set_metadata(exif_data) def apply_filter(self, filter_name): """ Wendet den ausgewählten Filter auf die Kamera an. """ filter_option = self.filters[filter_name] if filter_option is None: self.camera.image_effect = "none" else: self.camera.image_effect = filter_option def check_nas_availability(self, nas_address): """ Überprüft die Verfügbarkeit des NAS-Servers. """ try: subprocess.check_output(["ping", "-c", "1", nas_address]) # Beispiel für Linux return True except subprocess.CalledProcessError: return False def mount_nas(self, nas_address, local_mount_path): """ Mountet das NAS-Verzeichnis. """ subprocess.run(["mount", "-t", "cifs", f"//{nas_address}/share", local_mount_path]) # Der Befehl oben ist ein Beispiel für ein CIFS-Share, bitte anpassen def upload_pending(self, nas_directory): """ Lädt ausstehende Videodateien auf den NAS-Server hoch und entfernt sie danach. """ if self.check_nas_availability(self.nas_address): for source_path in self.pending_upload: filename = os.path.basename(source_path) video_date = filename.split("_")[1] # Extrahieren des Datums aus dem Dateinamen target_directory = os.path.join(nas_directory, video_date) self.create_directory(target_directory) # Verzeichnis auf NAS erstellen self.upload_video(source_path, target_directory) # Video kopieren und entfernen if self.delete_after_upload: os.remove(source_path) # Datei löschen nach erfolgreichem Upload self.pending_upload.remove(source_path) # Aus der Warteschlange entfernen else: print("NAS Server ist nicht verfügbar.") def add_to_pending_upload(self, source_path): """ Fügt eine Videodatei zur Warteschlange für das Hochladen hinzu. """ self.pending_upload.append(source_path) def upload_video(self, source_path, target_directory): """ Kopiert eine Videodatei auf den NAS-Server und entfernt sie danach. """ filename = os.path.basename(source_path) target_path = os.path.join(target_directory, filename) if not os.path.exists(target_path): shutil.copy2(source_path, target_path) # Kopieren der Datei os.remove(source_path) # Entfernen der Datei von der Kamera def create_directory(self, directory_path): """ Erstellt das Verzeichnis, wenn es nicht existiert. """ if not os.path.exists(directory_path): os.makedirs(directory_path) def scan_wifi(self): """ Scannt verfügbare WLAN-Netzwerke und zeigt das stärkste Signal auf dem Display an. """ wifi_scan_result = subprocess.check_output(["sudo", "iwlist", "wlan0", "scan"], universal_newlines=True) networks = self.parse_wifi_scan_result(wifi_scan_result) if networks: strongest_network = max(networks, key=lambda x: x["SignalStrength"]) self.display_wifi_info(strongest_network) self.current_menu.extend(networks) # Füge die Netzwerke dem Menü hinzu self.current_menu_index = len(self.menu) # Setze den Index auf das erste Netzwerk self.display_menu() else: self.display_message("Keine WLAN-Netzwerke gefunden.") def parse_wifi_scan_result(self, scan_result): """ Analysiert den Scan-Ergebnis-Text und gibt eine Liste von WLAN-Netzwerken zurück. """ networks = [] lines = scan_result.split("\n") network = {} for line in lines: if "Cell" in line: if network: networks.append(network) network = {} elif "ESSID" in line: network["SSID"] = line.split(":")[1].strip().strip("\"") elif "Signal level" in line: signal_strength = int(line.split("=")[1].split()[0]) network["SignalStrength"] = signal_strength return networks def display_wifi_info(self, network): """ Zeigt die WLAN-Informationen auf dem Display an. """ # Hier implementieren Sie den Code, um die WLAN-Informationen anzuzeigen. pass def send_deauth_packet(self, bssid, client_mac, count=1): """ Sendet Deauth-Pakete an einen Client oder Access Point. """ deauth_packet = RadioTap() / Dot11(type=0, subtype=12, addr1=client_mac, addr2=bssid, addr3=bssid) / Dot11Deauth() for _ in range(count): sendp(deauth_packet, iface="wlan0", verbose=False) def record_handshake(self, bssid): """ Aufzeichnung eines WLAN-Handshakes. """ #bssid = "Ziel_BSSID" #client_mac = "Client_MAC" handshake_packets = [] def packet_handler(pkt): if pkt.haslayer(Dot11): #if pkt.addr1 == client_mac and pkt.addr2 == bssid: if pkt.addr2 == bssid: handshake_packets.append(pkt) if len(handshake_packets) >= 4: self.save_handshake_packets(handshake_packets) print("Handshake aufgezeichnet!") return True sniff(iface="wlan0", prn=packet_handler, count=0) def save_handshake_packets(self, packets): """ Speichert die aufgezeichneten Handshake-Pakete in eine Datei. """ with open("handshake.pcap", "wb") as f: for pkt in packets: f.write(pkt) def collect_client_macs(self): """ Sammelt MAC-Adressen von Geräten in der Nähe. """ client_macs = [] def packet_handler(pkt): if pkt.haslayer(Dot11): if pkt.type == 0 and pkt.subtype == 4: # Probe Request client_mac = pkt.addr2 if client_mac not in client_macs: client_macs.append(client_mac) #print("Neues Gerät erkannt:", client_mac) sniff(iface="wlan0mon", prn=packet_handler, count=0) return client_macs def check_passwords(self, handshake_filename, passwords_filename): """ Überprüft Passwörter aus einem Wörterbuch auf Übereinstimmung mit dem Handshake. """ with open(handshake_filename, "rb") as f: handshake_data = f.read() with open(passwords_filename, "r") as f: passwords = f.read().splitlines() for password in passwords: if self.check_password(handshake_data, password): print("Passwort gefunden:", password) return password # Passwort gefunden, beenden Sie die Schleife print("Kein übereinstimmendes Passwort gefunden.") return None def check_password(self, handshake_data, password): """ Überprüft ein Passwort auf Übereinstimmung mit dem Handshake. """ # Berechnen Sie den Hash des Passworts und vergleichen Sie mit dem Handshake hash_calculated = hashlib.pbkdf2_hmac("sha1", password.encode(), ssid.encode(), 4096, 32) pmk = self.calculate_pmk(hash_calculated) ptk = self.calculate_ptk(handshake_data, pmk) calculated_mic = self.calculate_mic(ptk, handshake_data) # Vergleichen Sie den berechneten MIC mit dem MIC im Handshake # Hier hängt es von der Struktur des Handshake-Datenformats ab if calculated_mic == handshake_mic: return True else: return False def calculate_pmk(self, password): """ Berechnet den Pairwise Master Key (PMK) aus dem Passwort. """ return hashlib.pbkdf2_hmac("sha1", password.encode("utf-8"), ssid.encode("utf-8"), 4096, 32) def calculate_ptk(self, handshake_data, pmk): """ Berechnet den Pairwise Transient Key (PTK) aus dem Handshake und dem PMK. """ # Hier benötigen Sie den genauen Algorithmus zur Berechnung des PTK # Basierend auf dem Handshake-Datenformat und dem PMK pass def calculate_mic(self, ptk, handshake_data): """ Berechnet den Message Integrity Code (MIC) aus dem PTK und dem Handshake. """ # Hier benötigen Sie den genauen Algorithmus zur Berechnung des MIC # Basierend auf dem PTK und dem Handshake-Datenformat pass def run(self): """ Startet die Threads für Aufnahme, Display und Menü und lässt die Anwendung laufen. """ nas_address = "NAS_IP_Adresse" # Hier die tatsächliche IP-Adresse des NAS eintragen local_mount_path = "/pfad/zum/lokalen/mount" # Lokaler Pfad zum Mounten des NAS self.pending_upload = [] # Liste der Videodateien, die auf das Hochladen warten record_thread = threading.Thread(target=self.record_thread) # Aufnahme-Thread display_thread = threading.Thread(target=self.display_thread) # Display-Thread menu_thread = threading.Thread(target=self.menu_thread) # Menü-Thread record_thread.daemon = True display_thread.daemon = True menu_thread.daemon = True record_thread.start() display_thread.start() menu_thread.start() try: while True: # Hauptprogrammschleife pass except KeyboardInterrupt: # Programm beenden, wenn eine Tastaturunterbrechung erfolgt GPIO.cleanup() # GPIO-Pins aufräumen if __name__ == "__main__": app = RaZero() app.run()