Posts by Dennis89

#!/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)

#!/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

VIDEO_PATH = Path("/home/ich-bins/motioneye/Camera1")
VIDEO_FORMAT = [".h264", ".mp4"]
RECORDING_TIME = 10


def record_video(camera, date_time):
    print("Start Video-Aufnahme -", date_time)
    video_file = VIDEO_PATH.joinpath(f"{date_time}{VIDEO_FORMAT[0]}")
    camera.start_recording(str(video_file))
    camera.wait_recording(RECORDING_TIME)
    camera.stop_recording()
    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: Path) -> None:
    """
    Videos werden zuerst übertragen und danach transkodiert
    """
    # 1. transfer
    h264_video = video_file.with_suffix(VIDEO_FORMAT[0])
    transfer(h264_video)
    print(f"Übertragung von {h264_video} abgeschlossen", flush=True)

    # 2. transcoding
    # with this line 'change_video_format' don't have the original suffix
    # mp4_video = video_file.with_suffix(VIDEO_FORMAT[1])
    # 
    change_video_format(video_file)
    print(f"Transkodierung abgeschlossen", flush=True)

    # 3. entfernen
    h264_entfernen(video_file)
    print(f"Original {h264_video} entfernt")


def h264_entfernen(video_file):
    video_file.unlink()


def main(process_pool):
    camera = picamera.PiCamera()
    camera.resolution = (1920, 1080)
    camera.framerate = 30
    camera.brightness = 65
    while True:
        date_time = datetime.now().strftime("%Y%m%d_%H%M%S")
        camera.annotate_text = date_time
        video_file = record_video(camera, date_time)
        date_time = datetime.now().strftime("%H:%M:%S")
        print("Ende Video-Aufnahme - ", date_time)
        process_pool.submit(task, video_file)
        sleep(10)


if __name__ == "__main__":
    pool = ProcessPoolExecutor(max_workers=4)
    main(pool)

#!/usr/bin/env python3


def main():
    ich_bin_ein_string = 'Bisschen Tet'
    print(ich_bin_ein_string)
    print(type(ich_bin_ein_string))
    ich_bin_kein_string = 47
    print(ich_bin_kein_string)
    print(type(ich_bin_kein_string))
    jetzt_bin_ich_ein_string = str(ich_bin_kein_string)
    print(jetzt_bin_ich_ein_string)
    print(type(jetzt_bin_ich_ein_string))
    print('ich_bin_kein_string')

if __name__ == '__main__':
    main()

#!/usr/bn/env python3

from datetime import datetime
from ftplib import FTP
from pathlib import Path
from subprocess import run
from threading import Thread
from time import sleep

import picamera

VIDEO_PATH = Path("/home/ich-bins/motioneye/Camera1")
VIDEO_FORMAT = [".h264", ".mp4"]
RECORDING_TIME = 10


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)
    video_file.unlink()
    print("Video-Aufnahme konvertiert -", datetime.now().strftime("%Y%m%d_%H%M%S"))


def record_video(camera, date_time):
    print("Start Video-Aufnahme -", date_time)
    video_file = VIDEO_PATH.joinpath(f"{date_time}{VIDEO_FORMAT[0]}")
    camera.start_recording(str(video_file))
    camera.wait_recording(RECORDING_TIME)
    camera.stop_recording()
    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 main():
    camera = picamera.PiCamera()
    camera.resolution = (1920, 1080)
    camera.framerate = 30
    camera.brightness = 65
    while True:
        date_time = datetime.now().strftime("%Y%m%d_%H%M%S")
        camera.annotate_text = date_time
        video_file = record_video(camera, date_time)
        print("Ende Video-Aufnahme - ", datetime.now().strftime("%Y%m%d_%H%M%S"))
        transfer(video_file)
        print("h264-Transfer beendet - ", datetime.now().strftime("%Y%m%d_%H%M%S"))
        Thread(target=change_video_format, args=[video_file]).start()
        sleep(RECORDING_TIME)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3

from signal import pause

from RPi import GPIO as gpio

SENSOR_GPIO = 17


def check_state():
    if gpio.input(SENSOR_GPIO):
        print("Is closed")
    else:
        print("Is opened")


def main():
    try:
        gpio.setmode(gpio.BCM)
        gpio.setup(SENSOR_GPIO, gpio.IN, pull_up_down=gpio.PUD_UP)
        gpio.add_event_detect(SENSOR_GPIO, gpio.BOTH, callback=check_state())
        pause()
    except KeyboardInterrupt:
        gpio.cleanup()
        print("Bye!")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3


from gpiozero import DigitalInputDevice

SENSOR_GPIO = 17


def main():
    try:
        cover_sensor = DigitalInputDevice(SENSOR_GPIO)
        while True:
            if cover_sensor.is_active:
                print("Is closed")
                cover_sensor.wait_for_inactive()
            else:
                print("Is open")
                cover_sensor.wait_for_active()
    except KeyboardInterrupt:
        print("Bye!")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3


from threading import Thread
from time import monotonic, sleep

from gpiozero import Button, Device
from gpiozero.pins.mock import MockFactory

TIME_TO_WAIT = 10

#
# Line 13 to 40 is just to simulate the GPIO-Pin
#


class SensorMotion(Thread):
    def __init__(self, sensor):
        Thread.__init__(self)
        self.sensor = sensor
        self.sensor_state = [self.sensor.pin.drive_low, self.sensor.pin.drive_high]
        self.choices = 'high', 'low', 'exit simulation'
        self.letter_to_choice = {choice[0]: choice for choice in self.choices}

    def format_as_option(self, text):
        return f'({text[0]}){text[1:]}'

    def print_sensor_state(self):
        return f'GPIO-Pin-State is: {self.sensor.value}'

    def run(self):
        prompt = '{}? : '.format(', '.join(map(self.format_as_option, self.choices)))
        while True:
            user_entry = self.letter_to_choice.get(input(prompt))
            if user_entry == 'high':
                self.sensor_state[1]()
                print(self.print_sensor_state())
            elif user_entry == 'low':
                self.sensor_state[0]()
                print(self.print_sensor_state())
            elif user_entry == 'exit simulation':
                self.sensor_state[1]()
                print('Schönen Gruß und auf Wiedersehn!')
                break


class WashMachine:
    def __init__(self):
        self.count = None

    def control_counting(self):
        if self.count:
            self.count = False
        sleep(0.5)
        self.count = True

    def count_seconds(self):
        start = monotonic()
        print('counting...')
        while True:
            delta = monotonic() - start
            if not self.count:
                print('stop counting')
                break
            elif delta >= TIME_TO_WAIT:
                self.count = False
                print('Wäsche ist fertig\n Enter für weitere Optionen')
                break


def send_message():
    print('Wäsche ist fertig')


def main():
    #
    # Line 78, 81 and 86 are just for simulate GPIO-Pin
    #
    Device.pin_factory = MockFactory()
    vib_sensor = Button(17, hold_time=10, pull_up=False)
    washing_machine = WashMachine()
    sensor_simulation = SensorMotion(vib_sensor)
    vib_sensor.when_pressed = lambda: washing_machine.control_counting()
    vib_sensor.when_released = lambda: Thread(
        target=washing_machine.count_seconds
    ).start()
    sensor_simulation.start()


if __name__ == '__main__':
    main()