ㅤ

Automate your workflow with simple microservice built in Python

having an own custom microservice that exposes one or multiple HTTP endpoints is very useful when taking some actions (executing tasks) or serving data from database. As a DevOps engineers we are always trying to automate repeatable tasks in order to achieve the end result, improve productivity and remove human errors to the absolute minimum. Such services are quite popular in an event driven architectures when you want to run specific workload based on an event (HTTP call, new Pull request, logon failure, etc.)

examples how to use RESTful service

  • initiate task (for example some batch processing task)
  • invoke build of another application
  • deploy an infrastructure
  • run custom script on number of remote servers
  • scale your application running in the cloud environment
  • anything that can be triggered by an event and automate your workload

In my example I will show you how to setup RESTful service that plays my favorite online radio stations by executing radio player from command line. I’m using this application almost every day on my RaspberryPI 4 device, that is connected to my audio sound system.

What’s needed to build an API

I’m using Python in version 3.7, but it should work on the never versions too, below there is a list of external modules used for the project:

fastapi==0.61.1
uvicorn==0.12.1
rq==1.5.2
redis==3.5.3

There are also 2 application that have to be installed on your system (in order to play streams from command line):

  • mpg123
  • ffplay

why do I need 2 different applications? Well mpg123 is very lightweight but sometimes it’s not able to play streams with some specific urls, that’s why I also have ffplay as a backup application - unfortunately is slightly more resource hungry, but overall it’s not an issue.

FastAPI is relatively new web framework, it’s often considered as a better alternative to Flask, Uvicorn is webserver that will host FastAPI application. For this project I have also used Redis - which is in-memory database, and rq to utilize concept of queue and background tasks. (I know that it is also possible to do it in vanilla FastAPI, but redis and rq are well established and high performant modules designed specifically for such work, so I’m using that)

project structure:

.
├── app.py
├── config.py
├── env
├── logs
│   └── media_log.log
├── __pycache__
├── requirements.txt
└── tasks.py

our main file is app.py - where we have instantiated our FastAPI object and defined all the HTTP routes. Next config.py contains all necessary configuration - radio stations urls, and connection to our redis database, last but not least tasks.py contains several functions that will be executed in background (in separate process to be specific) whenever we hit certain http endpoint, those tasks are basically mixture of bash/shell scripts that can be executed by Python code - using subprocess module. Another reason that you might want to use background tasks is when those tasks take some time to be completed but at the same time you want to have immediate response to the HTTP request from your API.

our redis instance is running on localhost, but it could be running on completely different server if you will:

config.py:

from rq import Queue
import redis
import os

r = redis.Redis()
q = Queue(connection=r)

cwd = os.path.dirname(__file__)
media_log = os.path.join(cwd, "logs/media_log.log")

stations = {
    "antyradio": {
        "name":"Antyradio",
        "cmd": "ffplay https://an.cdn.eurozet.pl/ant-waw.mp3 -nodisp"
    },
    "chili-zet": {
        "name":"Chili ZET",
        "cmd": "ffplay https://ch.cdn.eurozet.pl/chi-net.mp3 -nodisp"
    },
    "radio-kampus": {
        "name":"Radio Kampus",
        "cmd": "mpg123 http://193.0.98.66:8005"
    },
    "radio-kolor": {
        "name":"Radio Kolor",
        "cmd": "ffplay https://s4.radio.co/s866693838/listen -nodisp"
    },
    "radio-eska": {
        "name":"Radio Eska",
        "cmd": "mpg123 http://waw01-01.ic.smcdn.pl:8000/2380-1.mp3"
    },
    "newonce-radio": {
        "name":"newonce.radio",
        "cmd": "ffplay https://streamer.radio.co/s93b51ccc1/listen -nodisp"
    },
    "zet-80": {
        "name":"Zet 80",
        "cmd": "ffplay https://zt.cdn.eurozet.pl/ZET080.mp3 -nodisp"
    },
    "zet-hits": {
        "name":"Zet Hits",
        "cmd": "ffplay https://zt.cdn.eurozet.pl/ZETHIT.mp3 -nodisp"
    }
}

config.py contains also path to logs/media_log.log file - which will hold information about current title song and Author that is played by radio station. I will create separate route in API to get data from that file.

tasks.py:

import subprocess


def stop_radio():
    kill_proc = subprocess.Popen(['pkill', '(mpg123|ffplay)'])
    kill_proc.communicate()


def play_station(cmd: str, app_name:str, media_log: str):
    stop_radio()
    if app_name == 'mpg123':
        subprocess.Popen(f"{cmd} > {media_log} 2>&1", shell=True)
    elif app_name == 'ffplay':
        subprocess.Popen(f"/bin/bash -c '{cmd} 2> >( stdbuf -oL awk /StreamTitle/ | tee {media_log} )'", shell=True)


def change_volume(volume: str):
    cmd = "amixer -c 0 | egrep '\[[0-9]+%\]' -o | cut -d '[' -f2 | cut -d '%' -f1"
    p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
    current_volume = int(p.stdout.read().decode('utf-8').strip())
    volume_option = {
        'up': current_volume +1,
        'down': current_volume -1,
        'unmute': 95,
        'mute': 0
    }
    set_option = volume_option.get(volume, None)
    cmd = f"amixer set -c 0 Headphone {set_option}%"
    subprocess.Popen(cmd, shell=True)

There are 3 functions in tasks.py: stop_radio(), play_station() and change_volume() They are combination of shell scripts, and all of them are executed and handled in separate process thanks to Redis and tasks queue functionality.

now let’s have a look at our main file:

app.py:

from fastapi import FastAPI, Response, status
from config import stations, media_log, q, r
from dataclasses import dataclass, field
import tasks
import subprocess
import re


@dataclass(frozen=True)
class Station:
    uri_name: str
    name: str = field(init=False)
    cmd: str = field(init=False)
    app_name: str = field(init=False)

    def __post_init__(self):
        try:
            object.__setattr__(self, 'name', stations[self.uri_name]["name"])
            object.__setattr__(self, 'cmd', stations[self.uri_name]["cmd"])
            object.__setattr__(self, 'app_name', self.cmd.split(' ')[0])
        except KeyError:
            object.__setattr__(self, 'name', None)
            object.__setattr__(self, 'cmd', None)
            object.__setattr__(self, 'app_name', None)


app = FastAPI()


@app.get('/play/{station}')
def play_station(station: str, response: Response):
    radio = Station(station)
    if radio.name is not None:
        q.enqueue(tasks.play_station, radio.cmd, radio.app_name, media_log)
        r.set('station', radio.name)
        return {'playing': radio.name}
    response.status_code = status.HTTP_404_NOT_FOUND
    return {'playing': False, 'station':'not found'}


@app.get('/change-volume/')
def change_volume(volume: str, response: Response):
    if volume in ['up','down','mute','unmute']:
        q.enqueue(tasks.change_volume, volume)
        return {'volume option': volume}
    response.status_code = status.HTTP_400_BAD_REQUEST
    return { 'valid query parameters': ['up','down','mute','unmute']}


@app.get('/stop')
def stop_radio():
    q.enqueue(tasks.stop_radio)
    return {'radio':'stopped'}


@app.get('/get-title')
def get_title():
    cmd = f"cat {media_log} | grep -a StreamTitle | tail -n1 | cut -d ';' -f1 | cut -d '=' -f2 | cut -d ':' -f2,3,4"
    p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
    title = p.stdout.read().decode('utf-8').strip()
    title = title.replace("';","'")
    station = r.get('station').decode('utf-8')
    if re.match('.*[a-zA-Z]+', title):
        return {'title': title, 'station': station}
    return {'title': 'unknown', 'station': station}

there is several routes defined in app.py and two of them accepts parameters:

  • /play/{station} - accept path parameter {station} - all station names are defined in config.py
  • /change-volume/ - accepts query parameters.

valid query parameters:

GET /change-volume/?volume=up
GET /change-volume/?volume=down
GET /change-volume/?volume=mute
GET /change-volume/?volume=unmute

What is really nice about FastAPI it the automatic documentation they provide, which requires zero configuration, you can use that documentation also to test your endpoints: more information about interactive documentation here image alt text

Is it stateless or stateful?

Now the application itself does not hold any state in the memory, but we have some functions that will output Stream information from radio station to the logfile: ’logs/media_log.log’, then another task uses shell script and Regular Expression to filter song title and ged rid of unnecessary information, another bit of information that is stored outside of application is the station name itself, that is stored in Redis.
/get-title endpoint receives all of that information and outputs in JSON format:

HTTP/1.1 200 OK
date: Fri, 18 Jun 2021 09:21:12 GMT
server: uvicorn
content-length: 84
content-type: application/json
Connection: close

{
  "title": "CatchUp - Weird Flex ft. Schafter prod. Karbid",
  "station": "newonce.radio"
}

Running uvicorn webserver

uvicorn is an ASGI webserver, that can be executed like this (it will run 4 instances of your web application on port 5000):

env/bin/uvicorn app:app --host 0.0.0.0 --port 5000 --workers 4 --log-level critical --no-access-log

Don’t forget to run rq worker to handle tasks:

env/bin/rq worker -q

Preparing HTTP requests:

Now to use that application it is necessary to prepare HTTP requests that you can fire up any time. On pc you can pre-define your urls for instance in Postman or VSCode and by clicking a button you can execute them. But in my case the most convenient way was to prepare HTTP calls on my Android phone. There is many different ways how to do it but I recommend really nice software for Android automation - It’s called Automate
This is how it looks on my mobile phone: image alt text

That’s it

Thanks if you’ve read that post till the end. It really means something to me :)
As always I’m sharing code in my github repository: https://github.com/Krzysi3k/RPI-FastApi-RadioPlayer Have a nice day!