Getting Started

Below is a python example on how to use the python bindings of the HyfeDetectionSDK to:

• Capture live audio from the microphone.
• Feed the audio to the SDK.
• Send the predictions obtained to the Insights API.

This tutorial is broken down into several sections with the full code provided at the end.

Setting up pyaudio

To use the SDK we need a source of audio samples. In this example pyaudio is used to capture live audio from the microphone. This function configures pyaudio to call the detect_cough callback function every 0.5 seconds. It also contains the main loop of the program, which checks if data is available to be synced to the Hyfe Insights api otherwise, it just sits and waits in sleep() while audio data is being feed to the SDK through the detect_cough callback.

def listen_and_detect():
    global pending_events

    sample_rate = 16000  # 16kHz

    # Start a pyAudio audio stream to obtain
    # live audio samples from the microphone
    p = pyaudio.PyAudio()
    stream = p.open(
        format=pyaudio.paFloat32,
        channels=1,
        rate=sample_rate,
        input=True,
        # half a second worth of audio samples
        frames_per_buffer=sample_rate * 0.5,
        stream_callback=detect_cough,
    )

    stream.start_stream()

    # Wait in a loop as detection runs
    while True:
        # Upload detected events to Hyfe Insights
        if len(pending_events) > 0:
            sync_data_to_hyfe_insights(pending_events)
            time.sleep(1)
        else:
            time.sleep(1)

Detecting cough (feeding audio to the SDK)

This is the function which was configured as a callback to pyaudio in the previous section.

To obtain predictions the audio samples are forwarded into the SDK using the detect() method. The predictions are stored in the pending_events array, which will be later used to sync the predictions to Hyfe Insights.

def detect_cough(in_data, frame_count, time_info, status):
    global pending_events

    # live microphone audio samples obtained from the microphone
    samples = numpy.frombuffer(in_data, "float32")

    # events detected by the SDK
    events_found = HyfeDetectionSDK.detect(samples=samples)

    # store this events so they can be uploaded later
    pending_events += events_found

    return (in_data, pyaudio.paContinue)

Sending predictions to the Insights API

This function takes the predictions obtained from the microphone live audio and sends them to the Insights API.

Each user has their own HYFE_UID and each device belonging to a user has a HYFE_DEVICE_ID. To post events a HYFE_API_KEY (provided by Hyfe) is also required.

Authorization via JWT is also supported but is not covered in this example.

HYFE_API_KEY = os.environ.get("HYFE_API_KEY")
HYFE_UID = os.environ.get("HYFE_UID")
HYFE_DEVICE_ID = os.environ.get("HYFE_DEVICE_ID")

def sync_data():
    global pending_events

    response = requests.post(
        "http://api.hyfe.ai/events/data",
        data=json.dumps(pending_events),
        headers={
            "X-API-KEY": HYFE_API_KEY,
            "HYFE-INSIGHTS-UID": HYFE_UID,
            "HYFE-INSIGHTS-DEVICE-ID": HYFE_DEVICE_ID,
        },
    )
    if response.status_code != 200:
        print(response.status_code, response.content)
        return False
    else:
        pending_events = []
        return True

Full code

Here is the full code used in this tutorial:

import os
import time
import pyaudio
import requests
import numpy
import json

import HyfeDetectionSDK

HYFE_API_KEY = os.environ.get("HYFE_API_KEY")
HYFE_UID = os.environ.get("HYFE_UID")
HYFE_DEVICE_ID = os.environ.get("HYFE_DEVICE_ID")

pending_events = []

def sync_data():
    global pending_events

    response = requests.post(
        "http://api.hyfe.ai/events/data",
        data=json.dumps(pending_events),
        headers={
            "X-API-KEY": HYFE_API_KEY,
            "HYFE-INSIGHTS-UID": HYFE_UID,
            "HYFE-INSIGHTS-DEVICE-ID": HYFE_DEVICE_ID,
        },
    )
    if response.status_code != 200:
        print(response.status_code, response.content)
        return False
    else:
        pending_events = []
        return True


def detect_cough(in_data, frame_count, time_info, status):
    global pending_events

    # live microphone audio samples obtained from the microphone
    samples = numpy.frombuffer(in_data, "float32")

    # events detected by the SDK
    events_found = HyfeDetectionSDK.detect(samples=samples)

    # store this events so they can be uploaded later
    pending_events += events_found

    return (in_data, pyaudio.paContinue)


def listen_and_detect():
    global pending_events

    sample_rate = 16000  # 16kHz

    # Start a pyAudio audio stream to obtain
    # live audio samples from the microphone
    p = pyaudio.PyAudio()
    stream = p.open(
        format=pyaudio.paFloat32,
        channels=1,
        rate=sample_rate,
        input=True,
        # half a second worth of audio samples
        frames_per_buffer=sample_rate * 0.5,
        stream_callback=detect_cough,
    )

    stream.start_stream()

    # Wait in a loop as detection runs
    while True:
        # Upload detected events to Hyfe Insights
        if len(pending_events) > 0:
            sync_data(pending_events)
            time.sleep(1)
        else:
            time.sleep(1)


if __name__ == "__main__":
    listen_and_detect()