Audiodateien (*.mp3) abspielen

Das folgende Aktorscript spielt die 8-Bit Chiptune Melodie von Mac Gyver ab. Folgende Setupschritte werden dafür benötigt. Das Setup basiert auf einem Raspberry OS (Debian 12 Bookworm) auf einem Raspberry Pi 3 B+.

cd /opt/fabinfra/adapters/
git clone https://gitlab.com/fabinfra/fabaccess/actors/python_process_template.git mp3play
cd /opt/fabinfra/adapters/mp3play/
python3 -m venv env
. env/bin/activate #activate venv

pip install pygame psutil

Wir müssen die Alsa-Audiokonfiguration anpassen und das Standardwiedergabegerät setzen

vim /etc/asound.conf

pcm.!default {
type asym
playback.pcm {
type plug
slave.pcm "hw:1,0"
}
}

cd /opt/fabinfra/adapters/mp3play/play_mp3.py

#!/opt/fabinfra/adapters/mp3play/env/bin/python3

import pygame

def play():
    print("In Use")
    file = '/opt/fabinfra/adapters/mp3play/8bit-macgyver.mp3'
    pygame.init()
    pygame.mixer.init()
    pygame.mixer.music.load(file)
    pygame.mixer.music.set_volume(1.0)
    pygame.mixer.music.play()

    while pygame.mixer.music.get_busy() == True:
         pass

if __name__ == "__main__":
    play()

cd /opt/fabinfra/adapters/mp3play/main.py

#!/opt/fabinfra/adapters/mp3play/env/bin/python3

import argparse
import psutil
import subprocess

def on_free(args, actor_name):
    PROCNAME = "play_mp3.py"
    for proc in psutil.process_iter():
        if PROCNAME in " ".join(proc.cmdline()):
            proc.kill()

def on_use(args, actor_name, user_id):
    cmd = "/opt/fabinfra/adapters/mp3play/env/bin/python3 /opt/fabinfra/adapters/mp3play/play_mp3.py"
    try:
        proc = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True)
    except OSError as e:
        raise OSError("{0}\nCommand failed: errno={1} {2}".format(' '.join(cmd), e.errno, e.strerror))

def on_tocheck(args, actor_name, user_id):
    print("To Check")

def on_blocked(args, actor_name, user_id):
    print("Blocked")

def on_disabled(args, actor_name):
    print("Disabled")

def on_reserve(args, actor_name, user_id):
    print("Reversed")

def on_raw(args, actor_name, data):
    print("Raw")


def main(args):
    new_state = args.state
    if new_state == "free":
        on_free(args, args.name)
    elif new_state == "inuse":
        on_use(args, args.name, args.userid)
    elif new_state == "tocheck":
        on_tocheck(args, args.name, args.userid)
    elif new_state == "blocked":
        on_blocked(args, args.name, args.userid)
    elif new_state == "disabled":
        on_disabled(args, args.name)
    elif new_state == "reserved":
        on_reserve(args, args.name, args.userid)
    elif new_state == "raw":
        on_raw(args, args.name, args.data)
    else:
        print("Process actor called with unknown state %s" % new_state)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("name", help="name of this actor as configured in bffh.dhall")

    subparsers = parser.add_subparsers(required=True, dest="state")

    parser_free = subparsers.add_parser("free")

    parser_inuse = subparsers.add_parser("inuse")
    parser_inuse.add_argument("userid", help="The user that is now using the machine")

    parser_tocheck = subparsers.add_parser("tocheck")
    parser_tocheck.add_argument("userid", help="The user that should go check the machine")

    parser_blocked = subparsers.add_parser("blocked")
    parser_blocked.add_argument("userid", help="The user that marked the machine as blocked")

    parser_disabled = subparsers.add_parser("disabled")

    parser_reserved = subparsers.add_parser("reserved")
    parser_reserved.add_argument("userid", help="The user that reserved the machine")

    parser_raw = subparsers.add_parser("raw")
    parser_raw.add_argument("data", help="Raw data for for this actor")

    args = parser.parse_args()
    main(args)