24.04.2025 Morsecode Converter mit FastApi

1. Python

Ein einfacher Morsecode übersetzer. Das Python-Skript liest eine Textdatei ein. Teilt diese anschließend in gleich große Stücke (Chunks) und sendet diese dann asynchron an einen FastAPI-Server. Der Client übersetzt die Chunks in Morsecode und schickt es dann zurück. Der Master empfängt die Fragmente, sortiert den Code korrekt und speichert dann das Ergebnis in einer Ausgabedatei.

1.1 Projektstruktur

  • master.py # Hauptskript für die Verarbeitung

  • client.py # Beinhaltet den MorseCode Converter-Teil

1.2 Befehl zum Ausführen

python master.py inputfile.txt outputfile.txt

  • inputfile.txt: Pfad zur Textdatei, die in Morsecode umgewandelt werden soll

  • outputfile.txt: Pfad zur Datei, in der das Ergebnis gespeichert wird

1.3 Code

1.3.1 master.py

Aufgaben:

  1. Einlesen der Textdatei

  2. Inhalt in kleine Blöcke (chunk_size = 100) teilen

  3. Weiterleiten an den Client zum Umwandeln

  4. Ergebnisse zusammenführen und in Ausgabedatei speichern

  5. Zeit für Verarbeitung ausgeben

Importierte Module:

  • time: zum Messen der Laufzeit

  • argparse: zur Verarbeitung von Kommandozeilenargumenten

  • asyncio: zur Verwaltung asynchroner Abläufe

  • aiohttp: für asynchrone HTTP-Requests (Client)

Text splitten:
def split_text_file(content, chunk_size=100):
    ```split text into smaller parts``` 
    return [content[i:i+chunk_size] for i in range(0, len(content), chunk_size)]

Splittet den Eingabetext in kleinere Teile und gibt eine Liste zurück

async def send_chunk(session, chunk, index):
    async with session.post(
        'http://127.0.0.1:8080/morse',
        json={'content': chunk, 'index': index}
    ) as response:
        data = await response.json()
        print(data)
        return data['index'], data['morse']

Sendet die Chunks an die Clients und wartet, bis alle Chunks bearbeitet wurden.

async def main():
    parser = argparse.ArgumentParser(description = "A morse code translator")
    parser.add_argument("input_file", help="Pfad zur Eingabedatei (Text)")
    parser.add_argument("output_file", help="Pfad zur Ausgabedatei (Morsecode)")

    args = parser.parse_args()

    try:
        with open(args.input_file, "r", encoding="utf-8") as infile:
            content = infile.read()
    except FileNotFoundError:
        print(f"Fehler: Datei '{args.input_file}' nicht gefunden.")
        return

Hier werden 2 Argumente (input_file und output_file) als Parameter in der Kommandozeile erwartet. Durch parser.parse_args() werden die Argumente ausgelesen und in args.infile bzw. args.output_file gespeichert. Anschließend wird die Eingabedatei versucht zu öffnen. Ist diese Vorhanden und kann ohne Fehler gelesen werden, wird der Inhalt in content gespeichert. Ansonsten wird ein Fehler ausgegeben.

    text_chunks = split_text_file(content)
    async with aiohttp.ClientSession() as session:
        tasks = [send_chunk(session, chunk, i) for i, chunk in enumerate(text_chunks)]
        results = await asyncio.gather(*tasks)

    results.sort(key=lambda x: x[0])
    morse_text = ''.join(morse for _, morse in results)
    with open(args.output_file, 'w') as output:
        output.write(morse_text)

Zuerst wird der eingelesene Text in kleinere Teile gespalten. (hier sind es 100 Zeichen pro Chunk/Teil). Dies wird benötigt um das Umwandeln zu parallelisieren. Anschließend wird eine asynchrone HTTP-Session gestartet (mit aiohttp). Innerhalb des blockes können HTTP-Request asynchron gesendet werden. Für jeden Chunk wird die Funktion send_chunk() aufgerufen. Der index i wird benötigt, um später die richtige Reihenfolge des Textes wieder herzustellen. Die tasks werden parallel ausgeführt. Durch asyncio.gather() wartet der Code, bis die POST-Requests beendet sind und speichert die Ergebnisse dann in results. Abschließend wird die results-Liste anhand der Indexe sortiert und durch „“.join in einem String gespeichert. Der Output wird in die angegebene Ausgabedatei gespeichert.

    start=time.time()

    end=time.time()
    total_time=end-start
    print("time:")
    print(str(total_time))

Hier wird die Zeit berechnet, die der Client zum Umwandeln des Textes braucht.

1.3.2 client.py

Aufgabe

Enthält die Logik für das Umwandeln von Text in Morsecode. Leerzeichen werden dabei als ‚|‘ umgewandelt.

Code
app = FastAPI()

Erstellt eine FastAPI-App-Instanz.

class MorseRequest(BaseModel):
    content: str
    index: int

  • content: der Textabschnitt der in Morsecode umgewandelt werden soll

  • index: Position im ursprünglichem Text


def encrypt(message):
    return ' '.join(MORSE_CODE_DICT.get(c.upper(), '|') for c in message)

encrypt konvertiert den übergebenen String in Morsecode.

@app.post("/morse")
async def convert_morse(data: MorseRequest):
    return {
        "index": data.index,
        "morse": encrypt(data.content)
    }

  • Endpunkt: POST /morse Gibt eine Response mit dem Index und dem übersetzten Morsecode zurück.

if __name__ == '__main__':
    uvicorn.run("client:app", host="0.0.0.0", port=8080, reload=True)

Wird ausgeführt durch python client.py.

2. Rust

An dieser Stelle wird die automatisch generierte Dokumentation von rustdoc verlinkt, da der Code in dieser Hinsicht ausreichend mit Docstrings versehen wurde:

2.1 client.rs

Client documentation

2.2 server.rs

Server documentation