Using smart lights for witness vote notifications

in HiveDevs4 years ago (edited)

I had been exploring the scripting options of Hue lights lately. One of the use cases of these smart lightbulbs is that you can actually use the lights and their colors for notifications. While playing with it, I've decided to create notifications on my witness' vote approvals/disapprovals.

The goal is simple:

  • Blink green when somebody approves my witness
  • Blink red when somebody disapproves my witness

To have the same setup, you need:

  • At least one Hue white and color smart lightbulb
  • Hue bridge

Getting Started


I've added three scenes to my target room (workspace).

  • Witness Vote Approval
  • Witness Vote Disapproval
  • Default

photo5843617973795928702.jpg

The software should listen to every transaction in the HIVE network, and if it sees an account_witness_vote operation targeted to our witness account, it should blink the lights green or red.

I would make the blinking colors in the scripting side, however, instead, I chose to trigger scenes so that I can play with the colors without touching the code again.

OperationAction
Witness ApprovalIMG-3061.gif
Witness DisapprovalIMG-3058.gif

Python script


import logging
import time

from lighthive.client import Client
from phue import Bridge
import threading
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# CONFIGURATION
HUE_BRIDGE_IP = "<ip>"  # can be shown at hue app
WITNESSES = ["emrebeyler",]
ROOM_NAME = "Workspace"
LIGHTBULB_NAME = "Workspace light"


class TransactionListener:

    def __init__(self, client, blockchain_mode="head",
                 start_block=None, end_block=None):
        self.client = client
        self.blockchain_mode = blockchain_mode or "irreversible"
        self.start_block = start_block
        self.end_block = end_block
        self.bridge = Bridge(HUE_BRIDGE_IP)


    def get_last_block_height(self):
        props = self.client.get_dynamic_global_properties()
        return props['head_block_number']

    def get_block(self, block_num):
        logger.info("Getting block: %s", block_num)
        block_data = self.client.get_block(block_num)
        return block_data

    def process_block(self, block_data):
        for transaction in block_data.get("transactions", []):
            for op_type, op_value in transaction.get("operations"):
                if op_type != "account_witness_vote":
                    continue

                if op_value.get("witness") not in WITNESSES:
                    continue

                t = threading.Thread(
                    target=self.notify,
                    args=(
                        op_value, ),
                    kwargs={},
                )
                t.start()

    def notify(self, op):

        default_state = self.bridge.get_light(LIGHTBULB_NAME)

        scene_name = 'Witness Vote Approval' if op.get("approve") else \
            'Witness Vote Disapproval'

        for i in range(0, 5):
            self.bridge.run_scene(ROOM_NAME, scene_name,
                        transition_time=5)
            time.sleep(0.5)
            self.bridge.run_scene(ROOM_NAME, 'Default', transition_time=5)
            time.sleep(0.5)

        defaults = {
            'on': default_state['state']['on'],
            'sat': default_state['state']['sat'],
            'hue': default_state['state']['hue'],
            'bri': default_state['state']['bri'],
        }

        self.bridge.set_light(LIGHTBULB_NAME, defaults, transitiontime=5)

    def listen(self):
        current_block = self.get_last_block_height()
        while True:
            while (self.get_last_block_height() - current_block) > 0:
                if self.end_block and current_block > self.end_block:
                    return
                else:
                    yield self.get_block(current_block)
                current_block += 1

            time.sleep(2.9)


if __name__ == '__main__':

    c = Client()
    tx_listener = TransactionListener(c)
    for block_data in tx_listener.listen():
        tx_listener.process_block(block_data)

Just update the configuration and run it. It requires lighthive and phue libraries installed in your system. I put this into supervisord so if somehow the script stops due to node failures, it should restart itself automatically.

Also, in a different Python file, you need to authenticate the phue library with your Hue bridge. Press the authenticate button in the Hue bridge and run this one:

from phue import Bridge
b = Bridge('<bridge_api>')
b.connect()

This is required only once.


A call to action, testing time! 😂


This lightbulb in the gifs is connected to my workspace's main lightning at the moment. If you didn't vote for my witness yet, vote for my witness on Hivesigner or on PeakD so that we can see the live result.

Currently testing is limited only to approval actions please don't troll me. :P

Cheers,

Sort:  

So theoretically if you make it blink only once per vote, it's possible to send some morse code through this? Using blinks as the pauses and the time in between as long/short presses.

you need to use time intervals of at least 3 secs then or can you do more than 1 vote in a block?

Min time intervals of 3 sec. So dots can be 3 sec wait, and dashes be 6 seconds.

but you can just do green light for short and red light for long actually, we are making things difficult here :-)

Yea, but my goal was to do it to @emrebeyler when he's running it.

Well, with a little bit of modification it should be possible. :)

So cool! I've got an idea for you regarding witnesses and polls, what's the best way to contact you for a chat? I'm @Brianoflondon pretty much everywhere.

thanks!

I'll dm you today or tomorrow.

It actually looks cool

Thanks! :)

Hi.. You know so much programming.. You are making different lights to glow for different actions..