consensus/engine.py at master · liveldp/consensus · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import consume
import annotation
from threading import Thread
from time import sleep
from datetime import datetime as dt, timedelta as delta
import calendar
import stats
from publish import publish
from store import r
import config
from api import start

consume.start_channel(config.AMQP_EXCHANGE, 'annotations.raw', 'consensus_input', annotation.annotation_callback)


def get_current_aggreement(fid, attr):
    return r.hget('f:aggr:{}'.format(fid), attr)


def set_current_aggreement(fid, attr, value):
    return r.hset('f:aggr:{}'.format(fid), attr, value)


def calculate_pollution(color, wattage, covered):
    if covered == 'false':
        return 'high'
    if color == 'blue':
        return 'high'
    elif color == 'yellow':
        if float(wattage) >= 100.0:
            return 'medium'
        return 'low'
    elif color == 'white':
        if float(wattage) >= 100.0:
            return 'high'
        return 'medium'

    return 'medium'


def check_annotations():
    while True:
        known_lampposts = annotation.get_lampposts()
        print '--------------------------------------- monitoring {} lampposts ---------------------------------------'.format(
            len(known_lampposts))
        for fid in known_lampposts:
            # print "> '{}':".format(fid)
            f_uri = r.get('f:uri:{}'.format(fid))
            agreements = stats.check_agreement(fid)
            agreed_color = None
            agreed_wattage = None
            agreed_cover = None
            position = annotation.get_lamppost_position(fid)
            position_dict = {}
            if position is not None:
                position_dict['latitude'] = position[0]
                position_dict['longitude'] = position[1]
            for agreement in agreements:
                agreement['id'] = fid
                attribute = agreement['attribute']
                if attribute == 'color':
                    agreed_color = agreement['value']
                if attribute == 'wattage':
                    agreed_wattage = agreement['value']
                if attribute == 'covered':
                    agreed_cover = agreement['value']
                value = get_current_aggreement(fid, attribute)
                if value != agreement['value']:
                    if f_uri is not None:
                        agreement['uri'] = f_uri
                    publish(dict(agreement, **position_dict))
                    set_current_aggreement(fid, attribute, agreement['value'])
                    annotation.delete_temporal(fid)
            try:
                if agreed_color and agreed_cover and agreed_wattage:
                    pollution = calculate_pollution(agreed_color, agreed_wattage, agreed_cover)
                    current_pollution = get_current_aggreement(fid, 'pollution')
                    if current_pollution is None or current_pollution != pollution:
                        set_current_aggreement(fid, 'pollution', pollution)
                        data = {'id': fid, 'attribute': 'pollution', 'value': str(pollution)}
                        if f_uri is not None:
                            data['uri'] = f_uri
                        publish(dict(data, **position_dict))
            except Exception as e:
                print e.message

        sleep(3)


monitor = Thread(target=check_annotations)
monitor.daemon = True
monitor.start()

start()

monitor.join()