Kannst mal versuchen ob der nachfolgende Code läuft. Ich habe noch nie etwas mit 'ProcessPool' gemacht und den Code jetzt auch nicht getestet.
Noch eine Anmerkung zu deiner Zeile 56. Wenn du in einer Funktion eine weitere aufrufen willst und dabei noch ein Argument 'water_led' übergeben willst, dann musst du dafür sorgen, das 'water_led' auch in der Funktion verfügbar ist. Sprich du musst es erst der Funktion übergeben, so wie mit 'video_file'.
Aber das ist ja eh der falsche Ort, denn es war ja mal gewünscht dass der Wasserwerfer und die Videoaufnahme parallel ablaufen.
Sollte mein Code nicht funktionieren, dann warten wir besser auf DeaD_EyE.
#!/usr/bin/env python3
from concurrent.futures import ProcessPoolExecutor
from datetime import datetime
from ftplib import FTP
from pathlib import Path
from subprocess import run
from time import sleep
import picamera
from gpiozero import LED, LightSensor, MotionSensor
VIDEO_PATH = Path("/home/ich-bins/motioneye/Camera1")
VIDEO_FORMAT = [".h264", ".mp4"]
RECORDING_TIME = 10
LIGHTSENSOR_PIN = 18
MOTION_SENSORS = [MotionSensor(23, queue_len=10), MotionSensor(24, queue_len=10)]
IRLED_PIN = 3
LED_PIN = 25
def start_watercannon(water_led):
print("Wasserwerfer an")
water_led.blink(on_time=1, off_time=0.5, n=1)
print("Wasserwerfer aus")
def record_video(camera, date_time, light_sensor, irled):
print("Video-Aufnahme läuft")
video_file = VIDEO_PATH.joinpath(f"{date_time}{VIDEO_FORMAT[0]}")
camera.start_recording(str(video_file))
if not light_sensor.light_detected:
irled.on()
print("Licht an")
camera.wait_recording(RECORDING_TIME)
camera.stop_recording()
irled.off()
return video_file
def transfer(video_file):
new_video_file = video_file.with_suffix(VIDEO_FORMAT[0])
with FTP('192.168.178.1', 'Ich-Bins', 'Password') as ftp, open(new_video_file, 'rb') as file:
ftp.storbinary(f'STOR {new_video_file.name}', file)
def change_video_format(video_file):
new_video_file = video_file.with_suffix(VIDEO_FORMAT[1])
run(["MP4Box", "-fps", "30", "-add", video_file, new_video_file], check=True)
def task(video_file) -> None:
"""
Videos werden zuerst übertragen und danach transkodiert
"""
transfer(video_file)
print(f"Übertragung von {video_file.name} abgeschlossen", flush=True)
change_video_format(video_file)
print(f"Transkodierung abgeschlossen", flush=True)
h264_entfernen(video_file)
print(f"Original {video_file.name} entfernt", flush=True)
def h264_entfernen(video_file):
video_file.unlink()
def check_exception(fut):
exc = fut.exception()
if exc:
print("Ausnahmefehler:", exc)
def main(process_pool):
camera = picamera.PiCamera()
camera.resolution = (1920, 1080)
camera.framerate = 30
camera.brightness = 65
water_led = LED(LED_PIN)
irled = LED(IRLED_PIN)
light_sensor = LightSensor(LIGHTSENSOR_PIN)
while True:
print('Warte auf Bewegung')
MOTION_SENSORS[0].wait_for_motion() and MOTION_SENSORS[1].wait_for_motion()
date_time = datetime.now().strftime("%Y%m%d_%H%M%S")
print("Bewegung erkannt - ", date_time)
camera.annotate_text = date_time
process_pool.submit(start_watercannon, water_led)
video_file = record_video(camera, date_time, light_sensor, irled)
print("Aufnahme beendet - ", datetime.now().strftime("%Y%m%d_%H%M%S"))
process_pool.submit(task, video_file)
future = process_pool.submit(task, video_file)
future.add_done_callback(check_exception)
print("Fertig für nächste Bewegungserkennung - ", datetime.now().strftime("%Y%m%d_%H%M%S"))
sleep(1)
if __name__ == "__main__":
pool = ProcessPoolExecutor(max_workers=4)
main(pool)
Alles anzeigen
Zu deinen Fehlermeldungen kann ich dir leider nichts sagen. Da müsste ich auch viel googln.
Grüße
Dennis