Nowoczesne aplikacje wymagają rozwiązań zapewniających łatwość ich utrzymania oraz skalowania. Takie możliwości może nam pomóc uzyskać Magazyn Zdarzeń. W artykule tym opowiem co to takiego, jakie korzyści nam daje i jak możemy go zastosować.

Artykuły na temat tworzenia aplikacji webowych.
Behat jest frameworkiem Behavior-driven development (w skrócie BDD) dla języka PHP, który pomaga w implementacji i testowaniu założeń biznesowych. Oficjalna dokumentacja Behata znajduje się na stronie https://docs.behat.org/.
Instalacja behata sprowadza się do zainstalowania paczki przy użyciu composera:
composer require --dev behat/behat
Następnie należy zainicjować behata, poleceniem:
$ vendor/bin/behat --init
Polecenie to stworzy katalog features i wypisze informacje na temat jego zawartości:
+d features - place your *.feature files here +d features/bootstrap - place your context classes here +f features/bootstrap/FeatureContext.php - place your definitions, transformations and hooks here
Podstawymi elementami Behata są pliki .feature oraz klasy kontekstu, powyższy rezultat pokazuje gdzie należy umieścić każde z nich.
Plik .feature jest to plik w formacie zwanym Gherkin lub Cucumber, opisujący dane rozwiązanie. Format opisu scenariuszy powinien być zrozumiały dla osób nietechnicznych. Na początku zawsze znajduje się opis danego rozwiązania, np.:
Feature: In order to buy products as a customer I have to put products into my basket and have enough money to pay for them.
Laravel Dusk jest narzędziem do automatyzowania przeglądarki i testowania stron budowanych w Laravelu. Narzędzie to uruchamia i steruje ChromeDriver, dzięki czemu mamy możliwość automatycznego testowania naszych aplikacji w przeglądarce.
Test driven development to technika tworzenia oprogramowania, która opiera się na cyklu trzech następujących po sobie czynności: test czerwony -> test zielony -> refactoring. Więcej informacji na temat TDD można znaleźć w sieci, ja natomiast polecam książkę „TDD. Techniki programowania sterowanego testami” Dariusza Woźniaka, kod tej książki jest pisany w C#, ale zawarte w niej informacje są uniwersalne dla wszystkich języków programowania.
Dla potrzeb naszego artykułu musimy o TDD wiedzieć tyle, że najpierw piszemy test który zakończy się niepowodzeniem, później piszemy kod który powoduje że test kończy się powodzeniem, następnie robimy refactoring i przechodzimy do pisania kolejnego testu, tym samym zapętlając cykl.
Któż nie lubi posłuchać radia, jedni słuchają dla muzyki, inni dla informacji, a jeszcze inni po prostu włączają radio jako tło. W dzisiejszych czasach nie jesteśmy już ograniczeni tylko do stacji radiowych, które odbiera nasz odbiornik fm, możemy słuchać audycji z całego świata. W niniejszym artykule pokażę jak zbudować swój własny odbiornik radiowy działający na mini-komputerze Raspberry Pi, sterowany przez przeglądarkę z dowolnego urządzenia w sieci wi-fi. Sercem systemu będzie aplikacja w języku Python.
Na początek wystarczy Raspberry Pi z systemem Raspbian i podłączone do niego słuchawki, na końcu artykułu przedstawię możliwości rozbudowy naszego radia.
Kod źródłowy napiszemy w języku Python 3, do stworzenia prostego interfejsu webowego wykorzystamy framework Flask, zaś za odtwarzanie dźwięku odpowiadać będzie aplikacja Music Player Daemon (mpd) oraz prosty interfejs konsolowy o nazwie mpc. Lista stacji radiowych będzie pobierana z pliku YAML.
Aby zainstalować wszystkie potrzebne narzędzia w systemie Ubuntu lub Raspbian należy wykonać polecenia:
sudo apt-get update sudo apt-get install mpd mpc python-pip pip install -U Flask oyaml
Stwórzmy katalog w którym będzie się znajdował nasz projekt a w nim plik main.py
.
W pliku main.py
utwórzmy klasę i stwórzmy jej obiekt:
class Radio: def __init__(self): None radio = Radio()
Następnie dodajemy metodę odpowiedzialną za komunikację z daemonem mpd:
def mpcCommand(self, cmd): p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE) return p.stdout.read()
Możemy już w tym momencie spróbować odtworzyć pierwszy stream dodając w metodzie __init__() kod:
self.mpcCommand(["mpc", "add", "http://bbcwssc.ic.llnwd.net/stream/bbcwssc_mp1_ws-einws"]) self.mpcCommand(["mpc", "play"])
Zaimportujemy też moduł subprocess:
import subprocess
Po wykonaniu aplikacji stream zacznie się odtwarzać, a aplikacja się zakończy. Aby zakończyć odtwarzanie musimy w konsoli systemowej wykonać polecenie mpc stop
.
Jeśli wykonamy polecenie mpc playlist
, zobaczymy że do naszej playlisty została dodana stacja „http://bbcwssc.ic.llnwd.net/stream/bbcwssc_mp1_ws-einws”. W celu uproszczenia komunikacji naszego skryptu z mpd, przed każdym rozpoczęciem odtwarzania będziemy czyścić listę i dodawać wybraną przez nas stację. Implementacja sprowadza się do dodania linii self.mpcCommand(["mpc", "clear"])
przed innymi wywołaniami self.mpcCommand()
. Dodatkowo na końcu naszej metody __init__(self) możemy dodać:
while True: None
Spowoduje to, że aplikacja będzie uruchomiona dopóki nie przerwiemy jej działania, na przykład kombinacją klawiszy CTRL+C. Przerwanie pracy aplikacji nie spowoduje jednak przerwania odtwarzania strumienia przez demon mpd, dlatego musimy dodać metodę exitHandler()
oraz zadeklarować jej wywołanie po zakończeniu aplikacji.
Dodajemy moduł atexit:
import atexit
oraz metodę:
def exitHandler(self): self.mpcCommand(["mpc", "stop"]) self.mpcCommand(["mpc", "clear"])
Jak widać metoda zatrzymuje odtwarzanie i czyści playlistę. Aby python wykonał tą metodę po zakończeniu aplikacji, należy metodzie __init__() dodać linijkę:
atexit.register(self.exitHandler)
Linia ta musi być umieszczona przed pętlą while True
.
Kolejnym krokiem będzie wczytanie listy stacji z pliku YAML. Przykładowy plik znajduje się tutaj, natomiast jego wczytanie wygląda następująco:
stationsFile = open("stations.yml", "r") data = yaml.load(stationsFile) self.stations = [] for k in data: self.stations.append({ "name": k, "url": data[k] })
Powyższy kod wczytuje zawartość pliku, ładuje go do zmiennej data
a następnie tworzy listę słowników self.stations
. Aby kod zadziałał musimy dodać jeszcze jeden moduł:
import oyaml as yaml
Użyłem oyaml zamiast PyYaml, ponieważ PyYaml nie zawsze wczytuje listę stacji w takiej kolejności jaką ustaliliśmy w pliku .yml, oyaml rozwiązuje ten problem.
Następnie usuwamy pętlę while True
a zamiast niej wstawiamy kod:
self.initWebUi()
Jest to wywołanie metody, której definicja wygląda następująco:
def initWebUi(self): app = Flask(__name__, template_folder="template") @app.route("/", methods=["GET", "POST"]) def control_page(title="Radio control"): if request.method == "POST": if request.form["submit"] == "Play": self.currentStationUrl = str(request.form["station"]) self.mpcCommand(["mpc", "clear"]) self.mpcCommand(["mpc", "add", self.currentStationUrl]) self.mpcCommand(["mpc", "play"]) elif request.form["submit"] == "Stop": self.mpcCommand(["mpc", "stop"]) return render_template("/control.html", title=title, stations=self.stations, currentStationUrl=self.currentStationUrl) app.run(host="0.0.0.0", port=1234)
Dodajemy też moduły:
from flask import Flask from flask import render_template from flask import request
Metoda initWebUi() inicjuje framework flask i definuje prosty routing wewnątrz którego jest akcja control_page
wyświetlająca szablon z pliku template/control.html
oraz obsługująca komendy „Play” i „Stop”. Plik szablonu znajduje się tutaj.
Dodatkowo w metodzie __init__() musimy dodać:
self.currentStationUrl = None
W tym momencie możemy otworzyć w przeglądarce adres http://0.0.0.0:1234/ i zobaczymy prosty panel pozwalający na wybór stacji i rozpoczęcie lub zatrzymanie odtwarzania.
Pełny kod aplikacji umiesciłem się w repozytorium na GitHub https://github.com/jakubthedeveloper/PythonInternetRadio, znajduje się w nim kod z artykułu z kilkoma usprawnieniami, na przykład możliwość podania w argumentach wywołania skryptu adresu i portu na którym ma działać nasze webowe UI. W pliku README.md znajdziemy również informację jak uruchomić projekt na mini-komputerze Raspberry Pi.
Dodatkowo kod w repozytorium został zrefaktoryzowany, poszczególne funkcje zostały przeniesione do odpowiednich klas, dodałem też unit testy.
W momencie gdy na słuchawkach podpiętych do Raspberry Pi słyszymy stację radiową wybraną w naszym webowym interfejsie mamy doskonały punkt wyjścia, żeby zbudować coś ciekawszego. Kilka pomysłów:
Zachęcam do eksperymentowania i podsyłania waszych realizacji.
Już po napisaniu powyższego artykułu dokonałem wielu modyfikacji, które znajdziecie w repozytorium projektu. Jedną z nich jest modyfikacja interfejsu webowego z wykorzystaniem Bootstrap i jQuery:
Moje radyjko zbudowane z wykorzystaniem rozwiązania przedstawionego w niniejszym artykule opiera się na następującej architekturze:
Załóżmy że mamy napisaną lub dopiero piszemy aplikację w PHP, która używa zewnętrznego API i z zwraca dane poprzez swoje własne API, czyli jest pośrednikiem.
Chcemy napisać testy naszych endpointów tak, żeby nie wykonywać zapytań do zewnętrznego API w czasie testów.
Jeden z naszych endpointów zwraca informacje o produkcie, jego adres to GET /product/{id}.
Nasz test mógłby wyglądać tak:
public function test_it_should_return_product() { $response = $this->get('/products/' . $this->faker->randomNumber()); $response->assertStatus(200); $response->assertJsonStructure([ "id", "name", "price" ]); }
W naszym kontrolerze używamy interfejsu ProductInterface który jest implementowany przez klasę ProductRepository, a ta z kolei używa klasy ProductProxy do wykonania zapytania do API. Klasas ProductProxy dziedziczy ogólną klasę BaseProxy, która zawiera metodę sendRequest(), to właśnie ta metoda odpowiada za wszelkie requesty HTTP do zewnętrznego API. Na początek więc zróbmy mock tej metody, aby upewnić się, że żaden z naszych testów nie wykona prawdziwego zapytania do zewnętrznego API.