Amazing Open Source Raspberry Pi Pico 2 Synth, Cutting Edge Power over Skin Research, and much more!

0:00:00 / -

Watch now on YouTube Watch now

Sad? Sing Along And Feel Better Instantly

Made by ghostrich / Art / Kids & Family / IoT

About the project

When detecting someone being sad, they are encouraged to sing a karaoke song to feel better. The AIY Vision Kit is being enhanced by using an ESP8266 as small display and MQTT to show the texts.

Project info

Difficulty: Moderate

Platforms: Arduino, Everything ESP, Raspberry Pi

Estimated time: 2 hours

License: GNU General Public License, version 3 or later (GPL3+)

Items used in this project

Hardware components

	Google Aiy Vision Full Kit - Includes Pi Zero Wh - V1.1	x 1
	ESP8266 incl. Display https://www.banggood.com/ESP8266-WIFI-Chip-0_91-Inch-OLED-CP2014-32Mb-Flash-Internet-Development-Board-p-1251877.html	x 1

Software apps and online services

Arduino IDE

Story

Why is this project interesting?

It is proven by science that singing makes you happy - it releases endorphins and reduces stress.

I have always been fascinated by how we can use technology to make people smile.

So if someone shows a sad face, they should be cheered up by singing a song.I was thinking that the Vision kit could get an easy upgrade by adding a small display to it.

The ESP8266 which includes a handy OLED display and Wifi is a practical device to use for this.

MQTT is used to send messages from the Raspberry to the micro controller and the micro controller shows the song texts.

How to set it up

On the Raspberry Pi:

Install the MQTT broker and client on the Raspberry Pi (Instructions: https://appcodelabs.com/introduction-to-iot-build-an-mqtt-server-using-raspberry-pi)

Go to the file "joy_detection_demo.py" (I attached the file as I am using it, but you will find it on the Raspberry Pi in this folder: /home/pi/AIY-projects-python/src/examples/vision/joy

Make sure to make a backup and keep the file in the same folder, so that it can be found by the respective service at startup and is called automatically.

There are two lines that need to be added to the file where the mood is detected:

os.system("mosquitto_pub -h localhost -t "aiy/message" -m "happy"")

and

os.system("mosquitto_pub -h localhost -t "aiy/message" -m "sad"")

You can test this by typing mosquitto_sub -t "aiy/message" on the console of the Raspberry Pi. You should see something like this, when you smile and frown and the camera:

On the ESP8266:

Open the attached "esp8266_mqtt_sing_a_song.ino" file in the Arduino IDE.

Install these libraries:

-- Adafruit SSD1306 (to use the display)

- - PubSubClient (for MQTT)

Add your wifi credentials and the IP address of the Raspberry Pi (you can use ifconfig on the console to find it out) at the top of the file.

Compile and upload it to the ESP8266.

To check, if everything is working, you can take a look at the Serial Monitor within the IDE. It should look something like this:

The End Result:

Here is a video of what happens, when a sad faces is detected (no sound):

Where can you go from here?

There are lots of interesting enhancements that can be done. Here are some ideas for you to try:

- add sound to play either from the Raspberry Pi or from the ESP8266

- play a different song every time someone is sad

- play a song when you detect happiness!

- find other ways to motivate the person with words

- invent a game moving an icon left or right depending on if the person is smiling or frowning

- invent a face gym!

- measure the time between smiling and frowning and show high scores

Code

joy_detection_demo.py

#!/usr/bin/env python3
# Copyright 2017 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Joy detection demo."""
import argparse
import collections
import contextlib
import io
import logging
import math
import os
import queue
import signal
import sys
import threading
import time

from PIL import Image, ImageDraw, ImageFont
from picamera import PiCamera

from aiy.board import Board
from aiy.leds import Color, Leds, Pattern, PrivacyLed
from aiy.toneplayer import TonePlayer
from aiy.vision.inference import CameraInference
from aiy.vision.models import face_detection
from aiy.vision.streaming.server import StreamingServer
from aiy.vision.streaming import svg

logger = logging.getLogger(__name__)

JOY_COLOR = (255, 70, 0)
SAD_COLOR = (0, 0, 64)

JOY_SCORE_HIGH = 0.85
JOY_SCORE_LOW = 0.10

JOY_SOUND = ('C5q', 'E5q', 'C6q')
SAD_SOUND = ('C6q', 'E5q', 'C5q')
MODEL_LOAD_SOUND = ('C6w', 'c6w', 'C6w')
BEEP_SOUND = ('E6q', 'C6q')

FONT_FILE = '/usr/share/fonts/truetype/freefont/FreeSans.ttf'

BUZZER_GPIO = 22

@contextlib.contextmanager
def stopwatch(message):
    try:
        logger.info('%s...', message)
        begin = time.monotonic()
        yield
    finally:
        end = time.monotonic()
        logger.info('%s done. (%fs)', message, end - begin)


def run_inference(num_frames, on_loaded):
    """Yields (faces, (frame_width, frame_height)) tuples."""
    with CameraInference(face_detection.model()) as inference:
        on_loaded()
        for result in inference.run(num_frames):
            yield face_detection.get_faces(result), (result.width, result.height)


def threshold_detector(low_threshold, high_threshold):
    """Yields 'low', 'high', and None events."""
    assert low_threshold < high_threshold

    event = None
    prev_score = 0.0
    while True:
        score = (yield event)
        if score > high_threshold > prev_score:
            event = 'high'
        elif score < low_threshold < prev_score:
            event = 'low'
        else:
            event = None
        prev_score = score


def moving_average(size):
    window = collections.deque(maxlen=size)
    window.append((yield 0.0))
    while True:
        window.append((yield sum(window) / len(window)))


def average_joy_score(faces):
    if faces:
        return sum(face.joy_score for face in faces) / len(faces)
    return 0.0


def draw_rectangle(draw, x0, y0, x1, y1, border, fill=None, outline=None):
    assert border % 2 == 1
    for i in range(-border // 2, border // 2 + 1):
        draw.rectangle((x0 + i, y0 + i, x1 - i, y1 - i), fill=fill, outline=outline)


def scale_bounding_box(bounding_box, scale_x, scale_y):
    x, y, w, h = bounding_box
    return (x * scale_x, y * scale_y, w * scale_x, h * scale_y)


def svg_overlay(faces, frame_size, joy_score):
    width, height = frame_size
    doc = svg.Svg(width=width, height=height)

    for face in faces:
        x, y, w, h = face.bounding_box
        doc.add(svg.Rect(x=int(x), y=int(y), width=int(w), height=int(h), rx=10, ry=10,
                         fill_opacity=0.3 * face.face_score,
                         style='fill:red;stroke:white;stroke-width:4px'))

        doc.add(svg.Text('Joy: %.2f' % face.joy_score, x=x, y=y - 10,
                         fill='red', font_size=30))

    doc.add(svg.Text('Faces: %d Avg. joy: %.2f' % (len(faces), joy_score),
            x=10, y=50, fill='red', font_size=40))
    return str(doc)


class Service:

    def __init__(self):
        self._requests = queue.Queue()
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def _run(self):
        while True:
            request = self._requests.get()
            if request is None:
                self.shutdown()
                break
            self.process(request)
            self._requests.task_done()

    def process(self, request):
        pass

    def shutdown(self):
        pass

    def submit(self, request):
        self._requests.put(request)

    def close(self):
        self._requests.put(None)
        self._thread.join()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, exc_tb):
        self.close()


class Player(Service):
    """Controls buzzer."""

    def __init__(self, gpio, bpm):
        super().__init__()
        self._toneplayer = TonePlayer(gpio, bpm)

    def process(self, sound):
        self._toneplayer.play(*sound)

    def play(self, sound):
        self.submit(sound)


class Photographer(Service):
    """Saves photographs to disk."""

    def __init__(self, format, folder):
        super().__init__()
        assert format in ('jpeg', 'bmp', 'png')

        self._font = ImageFont.truetype(FONT_FILE, size=25)
        self._faces = ([], (0, 0))
        self._format = format
        self._folder = folder

    def _make_filename(self, timestamp, annotated):
        path = '%s/%s_annotated.%s' if annotated else '%s/%s.%s'
        return os.path.expanduser(path % (self._folder, timestamp, self._format))

    def _draw_face(self, draw, face, scale_x, scale_y):
        x, y, width, height = scale_bounding_box(face.bounding_box, scale_x, scale_y)
        text = 'Joy: %.2f' % face.joy_score
        _, text_height = self._font.getsize(text)
        margin = 3
        bottom = y + height
        text_bottom = bottom + margin + text_height + margin
        draw_rectangle(draw, x, y, x + width, bottom, 3, outline='white')
        draw_rectangle(draw, x, bottom, x + width, text_bottom, 3, fill='white', outline='white')
        draw.text((x + 1 + margin, y + height + 1 + margin), text, font=self._font, fill='black')

    def process(self, message):
        if isinstance(message, tuple):
            self._faces = message
            return

        camera = message
        timestamp = time.strftime('%Y-%m-%d_%H.%M.%S')

        stream = io.BytesIO()
        with stopwatch('Taking photo'):
            camera.capture(stream, format=self._format, use_video_port=True)

        filename = self._make_filename(timestamp, annotated=False)
        with stopwatch('Saving original %s' % filename):
            stream.seek(0)
            with open(filename, 'wb') as file:
                file.write(stream.read())

        faces, (width, height) = self._faces
        if faces:
            filename = self._make_filename(timestamp, annotated=True)
            with stopwatch('Saving annotated %s' % filename):
                stream.seek(0)
                image = Image.open(stream)
                draw = ImageDraw.Draw(image)
                scale_x, scale_y = image.width / width, image.height / height
                for face in faces:
                    self._draw_face(draw, face, scale_x, scale_y)
                del draw
                image.save(filename)

    def update_faces(self, faces):
        self.submit(faces)

    def shoot(self, camera):
        self.submit(camera)


class Animator(Service):
    """Controls RGB LEDs."""

    def __init__(self, leds):
        super().__init__()
        self._leds = leds

    def process(self, joy_score):
        if joy_score > 0:
            self._leds.update(Leds.rgb_on(Color.blend(JOY_COLOR, SAD_COLOR, joy_score)))
        else:
            self._leds.update(Leds.rgb_off())

    def shutdown(self):
        self._leds.update(Leds.rgb_off())

    def update_joy_score(self, joy_score):
        self.submit(joy_score)


def joy_detector(num_frames, preview_alpha, image_format, image_folder,
                 enable_streaming, streaming_bitrate, mdns_name):
    done = threading.Event()
    def stop():
        logger.info('Stopping...')
        done.set()

    signal.signal(signal.SIGINT, lambda signum, frame: stop())
    signal.signal(signal.SIGTERM, lambda signum, frame: stop())

    logger.info('Starting...')
    with contextlib.ExitStack() as stack:
        leds = stack.enter_context(Leds())
        board = stack.enter_context(Board())
        player = stack.enter_context(Player(gpio=BUZZER_GPIO, bpm=10))
        photographer = stack.enter_context(Photographer(image_format, image_folder))
        animator = stack.enter_context(Animator(leds))
        # Forced sensor mode, 1640x1232, full FoV. See:
        # https://picamera.readthedocs.io/en/release-1.13/fov.html#sensor-modes
        # This is the resolution inference run on.
        # Use half of that for video streaming (820x616).
        camera = stack.enter_context(PiCamera(sensor_mode=4, resolution=(820, 616)))
        stack.enter_context(PrivacyLed(leds))

        server = None
        if enable_streaming:
            server = stack.enter_context(StreamingServer(camera, bitrate=streaming_bitrate,
                                                         mdns_name=mdns_name))

        def model_loaded():
            logger.info('Model loaded.')
            player.play(MODEL_LOAD_SOUND)

        def take_photo():
            logger.info('Button pressed.')
            player.play(BEEP_SOUND)
            photographer.shoot(camera)

        if preview_alpha > 0:
            camera.start_preview(alpha=preview_alpha)

        board.button.when_pressed = take_photo

        joy_moving_average = moving_average(10)
        joy_moving_average.send(None)  # Initialize.
        joy_threshold_detector = threshold_detector(JOY_SCORE_LOW, JOY_SCORE_HIGH)
        joy_threshold_detector.send(None)  # Initialize.
        for faces, frame_size in run_inference(num_frames, model_loaded):
            photographer.update_faces((faces, frame_size))
            joy_score = joy_moving_average.send(average_joy_score(faces))
            animator.update_joy_score(joy_score)
            event = joy_threshold_detector.send(joy_score)
            if event == 'high':
                logger.info('High joy detected.')
                os.system("mosquitto_pub -h localhost -t \"aiy/message\" -m \"happy\"")
                player.play(JOY_SOUND)
            elif event == 'low':
                logger.info('Low joy detected.')
                os.system("mosquitto_pub -h localhost -t \"aiy/message\" -m \"sad\"")
                player.play(SAD_SOUND)

            if server:
                server.send_overlay(svg_overlay(faces, frame_size, joy_score))

            if done.is_set():
                break

def preview_alpha(string):
    value = int(string)
    if value < 0 or value > 255:
        raise argparse.ArgumentTypeError('Must be in [0...255] range.')
    return value


def main():
    logging.basicConfig(level=logging.INFO)

    parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument('--num_frames', '-n', type=int, default=None,
                        help='Number of frames to run for')
    parser.add_argument('--preview_alpha', '-pa', type=preview_alpha, default=0,
                        help='Video preview overlay transparency (0-255)')
    parser.add_argument('--image_format', default='jpeg',
                        choices=('jpeg', 'bmp', 'png'),
                        help='Format of captured images')
    parser.add_argument('--image_folder', default='~/Pictures',
                        help='Folder to save captured images')
    parser.add_argument('--blink_on_error', default=False, action='store_true',
                        help='Blink red if error occurred')
    parser.add_argument('--enable_streaming', default=False, action='store_true',
                        help='Enable streaming server')
    parser.add_argument('--streaming_bitrate', type=int, default=1000000,
                        help='Streaming server video bitrate (kbps)')
    parser.add_argument('--mdns_name', default='',
                        help='Streaming server mDNS name')
    args = parser.parse_args()

    try:
        joy_detector(args.num_frames, args.preview_alpha, args.image_format, args.image_folder,
                     args.enable_streaming, args.streaming_bitrate, args.mdns_name)
    except KeyboardInterrupt:
        pass
    except Exception:
        logger.exception('Exception while running joy demo.')
        if args.blink_on_error:
            with Leds() as leds:
                leds.pattern = Pattern.blink(100)  # 10 Hz
                leds.update(Leds.rgb_pattern(Color.RED))
                time.sleep(1.0)

    return 0

if __name__ == '__main__':
    sys.exit(main())

esp8266_mqtt_sing_a_song.ino

#include <Adafruit_SSD1306.h>
#include <ESP8266WiFi.h>
#include <PubSubClient.h>

//set up mqtt client
const char* ssid = "WIFI SSID";
const char* password =  "WIFI Password";
const char* mqttServer = "Raspberry Pi IP address";
const int mqttPort = 1883;
const char* mqttUser = "aiy";
const char* mqttPassword = "message";

WiFiClient espClient;
PubSubClient client(espClient);

//set up screen
#define SCREEN_WIDTH 128 // OLED display width, in pixels
#define SCREEN_HEIGHT 32 // OLED display height, in pixels
#define OLED_RESET     -1 
Adafruit_SSD1306 display(SCREEN_WIDTH, SCREEN_HEIGHT, &Wire, OLED_RESET);

void setup() {
  Serial.begin(115200);
 
  WiFi.begin(ssid, password);
 
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.println("Connecting to WiFi..");
  }
  Serial.println("Connected to the WiFi network");
 
  client.setServer(mqttServer, mqttPort);
  client.setCallback(callback);
 
  connectMQTT();
 

  if(!display.begin(SSD1306_SWITCHCAPVCC, 0x3C)) {
    Serial.println(F("SSD1306 allocation failed"));
    for(;;);
  }
}

void connectMQTT() {
    while (!client.connected()) {
      Serial.println("Connecting to MQTT...");
 
      if (client.connect("ESP8266Client", mqttUser, mqttPassword )) {
        client.publish("aiy/message", "Hello from ESP8266");
        client.subscribe("aiy/message");
        Serial.println("connected");  
      } else {
        Serial.print("failed with state ");
        Serial.print(client.state());
        delay(2000);
    }
  }
}


void displaySongTextLine(String text, int textsize, int duration, int cursor_x, int cursor_y, int wait) {
  display.clearDisplay();
  display.display();
  display.setTextSize(textsize);
  display.setCursor(cursor_x,cursor_y);
  display.setCursor(0,0);
  display.println(text);
  display.display();
  delay(duration);
  display.clearDisplay();
  display.display();
  delay(wait);
}

void drawSongText(void) {
  display.clearDisplay();
  display.display();
  display.setTextSize(3);   
  display.setTextColor(SSD1306_WHITE);
  display.setCursor(30, 10); 
  display.cp437(true);      

  display.write(1);
  display.print("Sad?");
  display.display();
  delay(3000);
  display.clearDisplay();
  
  displaySongTextLine("Let's sing a song!\n\nThat'll cheer you up!",1,2000,10,10,2000);
  displaySongTextLine("If you're happy and\nyou know it clap\nyour hands...",1,2800,10,10,700);
  displaySongTextLine("If you're happy and\nyou know it clap\nyour hands...",1,2800,10,10,700);
  displaySongTextLine("If you're happy and\nyou know it...",1,1400,10,10,700);
  displaySongTextLine("and you really want\nto show it...",1,1400,10,10,700);
  displaySongTextLine("If you're happy and\nyou know it clap\nyour hands...",1,3000,10,10,2000);

  display.clearDisplay();
  display.display();
  display.setTextSize(3);
  display.setCursor(10, 5);
  display.write(2);
  display.print("Happy");
  display.display();
  delay(3000);
  display.clearDisplay();
  display.display();
}

void callback(char* topic, byte* payload, unsigned int length) {
 
  Serial.print("Message arrived in topic: ");
  Serial.println(topic);
  String recv_payload = String(( char *) payload);
 
  Serial.print("Message: ");

  if (recv_payload.startsWith("happy")) {  
    Serial.println("happy");
  } else  if (recv_payload.startsWith("sad")) {  
    Serial.println("sad");
    drawSongText();
  } else {
    //if a payload different from happy or sad is received
    Serial.println((char *)payload);
  } 
}

 
void loop() {
  connectMQTT();
  client.loop();
  delay(100);
}

Shop

Project Hub

Video

Blog

Shop home

3D printing

Single Board Computers

Development kits

Sensors

Internet of things

Robotics

Accessories

Product of the Week

Electromaker Educator

The Electromaker Show

The Electromaker Podcast

Sad? Sing Along And Feel Better Instantly

About the project

Project info