• Aleksandra Cynk
• Paweł Kopel
• Paweł Miziołek
• Sebastian Wąwoźny

Projekt za pomocą generatora, symulującego inteligentną opaskę, wysyła dane przy użyciu MQTT do AWS Topic. Dane te są przetwarzane za pomocą Lambdy i zapisywane do tabeli w DynamoDB. Ponadto w przypadku niepokojących danych(ryzyko śmierci itp.) wysyłane są ostrzeżenia na określony e-mail. Następnie kolejna lambda na podstawie aktualnych danych w tabeli dynamoDB tworzy plik csv ze wszystkimi danymi i umieszcza go na S3 Bucket. Plik ten jest wykorzystywany do generowania danych oraz wykresów za pomocą Pythona.

• Java
• Kotlin
• Python
• MQTT między opaską a backendem
• AWS IoT(IoT Core, DynamoDB, Lambda, S3)

Tutaj symulujemy opaskę. Przygotowane jest kilka „paczek” z parametrami odpowiadającymi np. treningowi, siedzeniu za biurkiem, spacerowi itp. Symulator losuje paczkę i czas do następnego losowania, następnie generuje losowe dane modulowane parametrami z paczki (np. trening - wyższe średnie tętno, siedzenie za biurkiem - niższe średnie tętno).

Paczki danych z urządzeń pomiarowych zostały stworzone na podstawie rzeczywistych urządzeń tego typu:

• smarthpone Google Nexus 5/5x
• smartphone Samsung Galaxy S5
• smartwatch LG G Watch

Użyty algorytm do liczenia kroków jest opisany w: "A More Reliable Step Counter using Built-in Accelerometer in Smartphone"

Generator potrafi też tworzyć alarmy. Symulują one naciśnięcie przycisku na opasce w trudnej sytuacji.

Przykładowe użycie generatora:

[user@host] $ ./gradlew run --console=plain
...
Starting generator
> state
activity: walking
sex: female
> display
> Measurement(uid=user2, time=2021-02-10T22:09:53.891657, temp=30.59574963431126, pulse=68.29102777760463, steps=0)
> Measurement(uid=user2, time=2021-02-10T22:09:55.892072, temp=31.979125339233804, pulse=68.29102777760463, steps=0)
> Measurement(uid=user2, time=2021-02-10T22:09:57.892483, temp=31.983099196585847, pulse=65.6949898861198, steps=1)
> Measurement(uid=user2, time=2021-02-10T22:09:59.892870, temp=31.44123789025402, pulse=65.6949898861198, steps=1) 
> Measurement(uid=user2, time=2021-02-10T22:10:01.893272, temp=31.720433496139083, pulse=65.6949898861198, steps=1)
> Measurement(uid=user2, time=2021-02-10T22:10:03.893635, temp=31.13546903824449, pulse=65.6949898861198, steps=1)
> stop
> alarm
published alarm
> exit
Done
...

Dane z generatora są wysyłane za pomocą MQTT Client na AWS Topic "/smartband"

Połączenie AWS używa klas z przykładu dołączonego do AWS SDK (w folderze java). Program wysyła pomiary do AWS-u w formacie JSON z poniższymi parametrami:

• uid (unique user id)
• czas pomiaru (urządzenie jest połączone do internetu więc możemy założyć istnienie RTC)
• kroki, liczone od zera w momencie uruchomienia urządzenia. Oprogramowanie po stronie AWS powinno wyliczać delty pomiędzy pomiarami.
• puls
• temperatura

private suspend fun publish() = delayLoop(5000, { input != "stop" }) {
    measurement.mutex.withLock {
        measurement.time = LocalDateTime.now(ZoneOffset.UTC).toString()
        GenConfigProvider.saveMeasurement(measurement)
        val msg = GenConfigProvider.serialize(measurement)
        aws.publish(topic, msg)
    }
}

Alarmy zawierają

• uid
• czas pomiaru
• puls
• temperatura
• "alarm"=true - informacja o typie wiadomości

private fun alarm() {
    val alarm = GenConfigProvider.alarmFromMeasurement(measurement)
    val msg = GenConfigProvider.serialize(alarm)
    aws.publish(topic, msg)
    println("published alarm")
}

Ustawiono Topic Rule, tak aby dane, które są wysyłane z generatora na topic "/smartband" były triggerem dla Lambda function.

W Lambda Function dane są przetwarzane i zapisywane do tabeli w DynamoDB za pomocą funkcji poniżej:

import boto3
import csv

def lambda_handler(event, context):
    client = boto3.client('dynamodb')
    response = client.put_item(
    TableName='SmartData',
    Item={
        'pulse':{'N':str(event['pulse'])},
        'steps':{'N':str(event['steps'])},
        'temp':{'N':str(event['temp'])},
        'time':{'S':event['time']},
        'uid':{'S':event['uid']}
    }
    )
    return 0

Utworzono także kolejną lambda function jako destination wcześniej utworzonej. Zadaniem nowej lambdy jest przetwarzanie danych z tabeli dynamoDB do pliku csv, który jest umieszczany na S3 bucket.

Kod przetwarzający dane z tabeli dynamoDB do pliku csv na S3 Bucket:

def lambda_handler(event, context):

    with open(TEMP_FILENAME, 'w') as output_file:
        writer = csv.writer(output_file)
        header = True
        first_page = True

        # Paginate results
        while True:
            # Scan DynamoDB table
            if first_page:
                response = table.scan()
                first_page = False
            else:
                response = table.scan(ExclusiveStartKey = response['LastEvaluatedKey'])

            for item in response['Items']:
                # Write header row?
                if header:
                    writer.writerow(item.keys())
                    header = False
                writer.writerow(item.values())

            # Last page?
            if 'LastEvaluatedKey' not in response:
                break

    # Upload temp file to S3
    s3_resource.Bucket(OUTPUT_BUCKET).upload_file(TEMP_FILENAME, OUTPUT_KEY)

Efektem tego kodu jest powstanie takiego pliku na S3 Bucket:

Z tego pliku bezpośrednio korzystamy do generowania szczegółowych danych oraz wykresów w pythonie.

Osobną lambdą (zawartą w folderze lambdas/mail sender) są odbierane alarmy z topicu. Na podstawie alarmów są wysyłane maile informujące np. operatora o potrzebie pomocy użytkownikowi.

Przykładowy mail:

Alarm send by user2 at 2021-02-09 17:12:15

Pulse: 87

Temperature: 32.57

Pobieramy dane z S3 Bucket za pomocą funkcji i przekształcamy je do listy słownikowej.

def open_csv() -> List[dict]:
    url = 'URL_TO_CSV_FILE_ON_S3_BUCKET'
    r = requests.get(url, allow_redirects=True)

    open('data.csv', 'wb').write(r.content)
    file = open('data.csv', 'r')
    return list(csv.DictReader(file))

Dane przetwarzamy i wyciągamy te aktualne i dla konkretnego użytkownika:

def read_csv():
    measurements = open_csv()
    measurements = get_for_day('2021-02-09', measurements)
    print(measurements)
    measurements = get_for_user('user2', measurements)

    print(steps_made_by_minmax(measurements))
    print(kcal(steps_made_by_minmax(measurements), 70))
    print('delta:' + str(steps_made_by_delta(measurements)))

    steps_chart(measurements, False, False)
    pulse_chart(measurements)

Na ich podstawie generujemy między innymi:

• Wykres odpowiadający pomiarowi pulsu
• Wykres odpowiadający pomiarowi kroków
• Liczba spalonych kcal
• Liczba kroków

Stworzyliśmy API restowe wystawiające endpointy dla opracowanych danych

opracowane dane zawierają:

• dzień, użytkownik lub brak (w zależności od endpointu, nie zwraca się danych podanych w zapytaniu i tworzących grupę)
• liczbę kroków
• liczbę spalonych kalorii