# Dec 17th, 2019: \[EN\] IoT: Christmas tree lights with Elasticsearch and Micro:bit

**URL:** <https://discuss.elastic.co/t/dec-17th-2019-en-iot-christmas-tree-lights-with-elasticsearch-and-micro-bit/211659>\
**Category:** Advent Calendar\
**Created:** [December 17, 2019, 9:00am UTC](https://discuss.elastic.co/t/dec-17th-2019-en-iot-christmas-tree-lights-with-elasticsearch-and-micro-bit/211659 "2019-12-17T09:00:06Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![Hendrik\_Muhs](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/hendrik_muhs/32/25802_2.png) [@Hendrik\_Muhs](https://discuss.elastic.co/u/Hendrik_Muhs)\
**Post date:** [December 17, 2019, 9:00am UTC](https://discuss.elastic.co/t/dec-17th-2019-en-iot-christmas-tree-lights-with-elasticsearch-and-micro-bit/211659/1 "2019-12-17T09:00:06Z")

</div>

We are getting closer to Christmas and it's time to setup and decorate the Christmas tree. Lights are an important part of it and as geek Elastician we decided to try something different this year.

**What if Christmas tree lights are based on data from Elasticsearch?**

The ingredients for this IoT fun project:

- [BBC micro:bit](https://en.wikipedia.org/wiki/Micro_Bit), a small ARM-based microcontroller
- [ESP 8266 wifi SoC](https://en.wikipedia.org/wiki/ESP8266)
- 1 programmable RGB strip (WS2812)
- breakout board for connecting all pieces, e.g. micro:Mate
- USB power bank or power plug
- some wires

 ![image](https://us1.discourse-cdn.com/elastic/original/3X/d/d/dd71dd115abcef2a3e926cfd0074d2a4ef3c5ffe.jpeg)

## Programming Language: Python

For programming the micro:bit we will use [MicroPython](https://en.wikipedia.org/wiki/MicroPython), the [micro:bit version](https://microbit-micropython.readthedocs.io/en/latest/index.html) is pre-installed and contains all we need, including a serial interface (uart) to talk to the wifi SoC and a library to program the lights (neopixel).

A breakout, also called expansion board, connects the different parts. The wifi SoC is connected to pin 0 (transmit) and pin 1 (receive), the led strip is connected to pin 2. All parts are powered over USB.

## Connecting to Wi-Fi

Let's start with connecting to the wifi using a serial interface between the micro:bit and the ESP 8266 SoC:

```auto
from microbit import uart

```

A small utility method helps us to read responses:

```auto
from microbit import Image, display, sleep
import utime

timeout = 10000 # 10 seconds
buffer = bytearray(64)
def readResponse():
    global buffer
    result = ""
    t = utime.ticks_ms()
    while utime.ticks_ms() < t + timeout:
        if uart.any():
            bytes = uart.readinto(buffer)
            result += str(buffer[:bytes], "utf-8")
     
            if '\r' in result:
                return result
            if len(result) > 500:
                return None
     
        sleep(50)
        
    # timeout :-(
    return None

```

You might notice that every response uses `\r` as a termination signal. As we will see below, every command we send uses \r as an end marker too.

Writing code for a microcontroller is different, often very low-level and has some challenges. One example – specific to micro:bit – is the 64 byte buffer which overwrites itself if you are not consuming the data fast enough. We do not want to get stuck if we miss the termination signal. A timeout of 10s ensures that we eventually return from `readResponse`. Another more general challenge is memory consumption: The heap space on a micro:bit is limited – the whole device has only 16kB RAM – therefore we must be careful and limit the maximum response size. 500 Bytes is an educated guess after some trial and error run.

But let’s now to the local wifi. We need to use another special [API](https://iot.dfrobot.com/docs/) to send the SSID and password in a sequence of `|`s and magic numbers:

```auto
def initWifiConnection():
    uart.write('\r')
    readUntil()
    t = utime.ticks_ms()
 
    uart.write('|2|1|wifi_ssid,wifi_pwd|\r')
    while utime.ticks_ms() < t + timeout:
        ip = readUntil('\r')
        if '|2|3|' in ip:
            return "WIFI OK"
    
    display.show(Image.ANGRY)
    return None

```

The expected response `|2|3|` signals a successful connection, let's connect:

```auto
uart.init(baudrate=9600, tx = pin0, rx = pin1)
if initWifiConnection() is None:
    display.show(Image.ANGRY)

```

Note that it takes a while until we have a connection. The SoC sends status updates after we initiated the wifi connection but only after receiving `|2|3|` do we have a connection and can continue. In case the wifi connection fails our micro:bit gets "angry". Using the built-in 25 pixel LED display is useful to see the state of execution, because you otherwise have no indication (while the serial interface is re-programmed to talk to the SoC USB debugging is off).

## Sending an HTTP Get request

The built-in HTTP client of our wifi SoC is triggered by sending `|3|1|` plus the endpoint:

```auto
def getRequest(endpoint):
    t = utime.ticks_ms()
    uart.write("|3|1|" + endpoint + "|\r")
    while utime.ticks_ms() < t + timeout:
        data = readResponse()
        if data is None:
            return None
        if '|3|200|' in data:
            return data[7:]
    display.show(Image.ANGRY)
    return None

```

If it works as expected the client returns `|3|` followed by the HTTP status code (`200` means success). We drop the response prefix and only return the data which comes next.

## Calling Elasticsearch

Let's now connect to Elasticsearch. The micro:bit python distribution lacks a lot of standard python libraries, `json` is one of them, so either we write some simple parsing or we use something different. The [cat APIs](https://www.elastic.co/guide/en/elasticsearch/reference/current/cat.html) are easy to parse and consume. We could get various health statuses or statistics. Let's use the [segments API](https://www.elastic.co/guide/en/elasticsearch/reference/current/cat-segments.html) for this project to retrieve doc counts:

```auto
endpoint = 'http://elasticsearch_ip:9200/_cat/segments/my_index?h=docs.count'

```

## Visualizing the data using the RGB strip

But the goal is not to show these numbers on the display, we want to use our RGB strip with 60 (neo-)pixels. Each pixel can be set to a different color. Let's try something similar to a heat map, we calculate a color from a data point based on the overall range of the data:

```auto
def rgb(minimum, maximum, values):
    minimum = float(minimum)
    maximum = float(maximum) + 0.01
    
    for value in values:
        ratio = 2 * (value-minimum) / (maximum - minimum) 
        b = int(max(0, 255*(1 - ratio))) 
        r = int(max(0, 255*(ratio - 1))) 
        g = 255 - b - r 
        yield (r, g, b)

```

Red, green and blue: we have a full byte for every portion. In this scheme high numbers have a bigger red part, low numbers get a bigger blue portion, green is somewhat in the middle.

In order to avoid hardcoding a minimum and a maximum we adapt the color scheme on every iteration. The generated output is a triplet of color values. That matches the format we need for the RGB LED strip using the [neopixel](https://microbit-micropython.readthedocs.io/en/latest/neopixel.html) module:

```auto
import neopixel
neopixels = 60
np = neopixel.NeoPixel(pin2, neopixels)

```

> Setting pixels works like accessing elements by index in python lists, e.g. for setting the 5th pixel we do:
> 
> ```auto
> np[4] = (255,0,0)
> 
> ```

Finally to activate the new colors, we call `np.show()`.

## Putting it together

Almost there! We have wifi access, we can do HTTP requests and we can visualize numeric values. Time to put it all together with a loop that regularly calls our endpoint, turns the output into a list of colors and sends them to the RGB strip:

```auto
while True:
    data = getRequest(endpoint)
    if data is not None and len(data) > 10:
        counts = list(map(toInt, data.split('\n')[:-1]))
        counts = counts[:neopixels]
        i = 0
        for color in rgb(min(counts), max(counts), counts):
            np[i] = color
            i+=1
        for j in range(i, neopixels):
            np[j] = (0,0,0)
        del(counts)
        np.show()

    sleep(sleeptime)

```

We are done, using the segments API we get a nice visualization of how Lucene merges data behind the scenes:

 ![led-strip](https://us1.discourse-cdn.com/elastic/original/3X/f/d/fda2fb2e91a2874e3c439edefd18c5cef60df093.gif)

Other ideas could be the state of individual nodes, cluster load, alerts, ML anomalies or incoming data rate. A nice alternative to dashboards on wall mounted screens.

## Final thoughts

IoT devices are fun, in this little project we created Christmas tree lighting driven by data received from Elasticsearch. However, most applications are the other way around: sending sensor data off to a data platform. At scale this would be data from many devices (via [MQTT](https://en.wikipedia.org/wiki/MQTT)). All the data must be processed and analyzed at scale, no problem with Elasticsearch.

The full program:

```auto
from microbit import pin0, pin1, pin2, Image, uart, display, sleep
import neopixel
import utime

timeout = 10000 # 10 seconds

def toInt(x):
    try:
        return int(x)
    except:
        return 0

def rgb(minimum, maximum, values):
    minimum = float(minimum)
    maximum = float(maximum) + 0.01
    
    for value in values:
        ratio = 2 * (value-minimum) / (maximum - minimum) 
        b = int(max(0, 255*(1 - ratio))) 
        r = int(max(0, 255*(ratio - 1))) 
        g = 255 - b - r 
        yield (r, g, b)

buffer = bytearray(64)
def readResponse():
    global buffer
    result = ""
    t = utime.ticks_ms()
    while utime.ticks_ms() < t + timeout:
        if uart.any():
            display.show(Image.HAPPY)
            
            bytes = uart.readinto(buffer)
            result += str(buffer[:bytes], "utf-8")
     
            if '\r' in result:
                return result
            if len(result) > 500:
                display.show(Image.ANGRY)
                return None
     
        sleep(50)
        
    # timeout :-(
    display.show(Image.ANGRY)
    return None

def initWifiConnection():
    uart.write('\r')
    readResponse()
    t = utime.ticks_ms()
 
    uart.write('|2|1|wifi_ssid,wifi_pwd|\r')
    while utime.ticks_ms() < t + timeout:
        ip = readResponse()
        if '|2|3|' in ip:
            return "WIFI OK"
    
    display.show(Image.ANGRY)
    return None

def getRequest(endpoint):
    t = utime.ticks_ms()
    uart.write("|3|1|" + endpoint + "|\r")
    while utime.ticks_ms() < t + timeout:
        data = readResponse()
        if data is None:
            return None
        if '|3|200|' in data:
            return data[7:]
    display.show(Image.ANGRY)
    return None

neopixels = 60
np = neopixel.NeoPixel(pin2, neopixels)

for i in range(0, neopixels):
    np[i] = (40,40,40)
np.show()

sleeptime = 50
endpoint = 'http://192.168.0.150:9200/_cat/segments/my_index?h=docs.count&s=docs.count'
#endpoint = 'http://192.168.0.150:9200/_cat/segments/my_index?h=docs.count'

uart.init(baudrate=9600, tx = pin0, rx = pin1)
if initWifiConnection() is None:
    display.show(Image.ANGRY)

np.clear()

while True:
    data = getRequest(endpoint)
    if data is not None and len(data) > 10:
        counts = list(map(toInt, data.split('\n')[:-1]))
        counts.reverse()
        counts = counts[:neopixels]
        i = 0
        for color in rgb(min(counts), max(counts), counts):
            np[i] = color
            i+=1
        for j in range(i, neopixels):
            np[j] = (0,0,0)
        del(counts)
        np.show()

    sleep(sleeptime)

```

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [November 4, 2022, 7:37am UTC](https://discuss.elastic.co/t/dec-17th-2019-en-iot-christmas-tree-lights-with-elasticsearch-and-micro-bit/211659/8 "2022-11-04T07:37:15Z")

</div>