Ben
Published © MIT

Air Pollution Monitor

Proof of concept of a Tangle based transparent and trusted air pollution monitoring system for future smart districts.

BeginnerFull instructions provided1 hour2,491
Air Pollution Monitor

Things used in this project

Story

Read more

Schematics

Air Pollution Monitor Architecture and Components

Code

pip_requirements.txt

Plain text
asn1crypto==0.24.0
astroid==2.2.5
backcall==0.1.0
certifi==2019.3.9
cffi==1.12.2
chardet==3.0.4
class-registry==2.1.2
Click==7.0
cryptography==2.6.1
decorator==4.4.0
enum-compat==0.0.2
filters==1.3.2
idna==2.8
ipython==7.4.0
ipython-genutils==0.2.0
isort==4.3.16
itsdangerous==1.1.0
jedi==0.13.3
Jinja2==2.10
lazy-object-proxy==1.3.1
MarkupSafe==1.1.1
mccabe==0.6.1
parso==0.3.4
pexpect==4.6.0
pickleshare==0.7.5
pkg-resources==0.0.0
prompt-toolkit==2.0.9
ptyprocess==0.6.0
pycparser==2.19
pygatt==3.2.0
Pygments==2.3.1
pylint==2.3.1
pyOpenSSL==19.0.0
PyOTA==2.0.7
PyOTA-CCurl==1.0.9
pyserial==3.4
pysha3==1.0.2
python-dateutil==2.8.0
pytz==2018.9
regex==2019.3.12
requests==2.21.0
six==1.12.0
traitlets==4.3.2
typed-ast==1.3.1
urllib3==1.24.1
wcwidth==0.1.7
Werkzeug==0.15.1
wrapt==1.11.1

APM_IOTA_firmware_F.bin

Plain text
APM Firmware: Disclaimer: This binary is provided "as is" and "with all faults", the author is not responsible of any damage caused to the device.
No preview (download only).

iota_create_address.py

Python
Simple IOTA Address Generator
""" IOTA Simple Address Generator"""
# coding: utf-8
from getpass import getpass
import iota

def main():
    """ IOTA simple address generator """

    print("Enter your seed and press \"Enter\" key")
    seed = getpass('Seed: ')
    generator = iota.crypto.addresses.AddressGenerator(seed)

    print('Enter the index of the address [default: 0]:')
    idx = input()

    try:
        if int(idx) >= 0:
            idx = int(idx)
        else:
            idx = 0
    except ValueError:
        idx = 0

    addr = generator.get_addresses(start=idx, count=1)
    print(addr[0])

if __name__ == "__main__":
    main()

iota_get_info.py

Python
Gets Messages from a specific IOTA Address
""" IOTA get messages """
# coding: utf-8
from iota import Iota
from iota.commands.extended.utils import find_transaction_objects

def main():
    """ iota get messages """
    addr = "WQVHGZMNXUBTR99WQRAQJCOCNHYUTNTI9DMDZKXITVVEFFFSFJTFXJNKVZGDGHFRSKLXTZCDRQJLQPO9W"
    addresses = [addr,]

    api = Iota("https://pool.trytes.eu", b'INSERTYOURSEEDHERE')

    transactions = find_transaction_objects(adapter=api.adapter, addresses=addresses)

    for transaction in transactions:
        message = transaction.signature_message_fragment
        print(message.decode())

if __name__ == "__main__":
    main()

iota_transfer_sensor_data.py

Python
Air Pollution Monitor main application
##!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" IOTA Test """
import time
import json

import iota
import pygatt

SENSOR_ADDRESS = "00:00:00:00:00:00" # Put here your sensor address

CHAR_UUIDS = {
    "tvoc" : "08d41e61-ac11-40a2-a95a-e0e0fb5336f9",
    "co2" : "08d41e61-ac11-40a2-a95a-e0e0fb5336fa",
}

def read_sensor():
    """ Read sensor values """
    measurements = []

    adapter = pygatt.GATTToolBackend()

    try:
        adapter.start()
        conn = adapter.connect(SENSOR_ADDRESS)

        time_info = int(time.time())

        # Get sensor values
        sensor_values = {}
        for sensor in CHAR_UUIDS:
            value = conn.char_read(CHAR_UUIDS[sensor])
            sensor_values[sensor] = float(value.decode("utf-8").rstrip('\x00'))

        # Prepare values to be transferred
        for sensor in sensor_values:
            measurement = {"measurement" : sensor, "time" : time_info, "value" : sensor_values[sensor]}
            measurements.append(measurement)

    except Exception as e:
        print("Error: ", e)

    finally:
        adapter.stop()
        return measurements


def transfer_to_iota():
    """ Transfers sensor data to the tangle"""
    seed = "INSERTYOURSEEDHERE"
    dest = "WQVHGZMNXUBTR99WQRAQJCOCNHYUTNTI9DMDZKXITVVEFFFSFJTFXJNKVZGDGHFRSKLXTZCDRQJLQPO9W"

    msg = read_sensor()
    print(msg)

    api = iota.Iota("https://potato.iotasalad.org:14265", seed)
    bundle = api.send_transfer(
        depth=3,
        transfers=[
            iota.ProposedTransaction(
                address=iota.Address(dest),
                value=0,
                tag=iota.Tag(b'SENSORDATA'),
                message=iota.TryteString.from_string(json.dumps(msg)),
                )])
    print("Transferred bundle hash: %s " %(bundle['bundle'].hash))

if __name__ == "__main__":
    transfer_to_iota()

Credits

Ben

Ben

4 projects • 10 followers

Comments