view service/powerEagle/reader.py @ 1562:c2ed44ed1e3c dependabot/pip/service/collector/twisted-19.7.0

Bump twisted from 19.2.0 to 19.7.0 in /service/collector Bumps [twisted](https://github.com/twisted/twisted) from 19.2.0 to 19.7.0. - [Release notes](https://github.com/twisted/twisted/releases) - [Changelog](https://github.com/twisted/twisted/blob/trunk/NEWS.rst) - [Commits](https://github.com/twisted/twisted/compare/twisted-19.2.0...twisted-19.7.0) Signed-off-by: dependabot[bot] <support@github.com>
author dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
date Fri, 14 Feb 2020 10:01:26 +0000
parents df80deeef113
children e8654a3bd1c7
line wrap: on
line source

#!bin/python
import json, time, os, binascii, traceback

from cyclone.httpclient import fetch
from docopt import docopt
from greplin import scales
from greplin.scales.cyclonehandler import StatsHandler
from influxdb import InfluxDBClient
from twisted.internet import reactor
from twisted.internet.defer import inlineCallbacks
import cyclone.web
from rdflib import Namespace, Literal

from standardservice.logsetup import log, verboseLogging
from patchablegraph import PatchableGraph, CycloneGraphHandler, CycloneGraphEventsHandler

from private_config import deviceIp, cloudId, installId, macId, periodSec
ROOM = Namespace("http://projects.bigasterisk.com/room/")

STATS = scales.collection('/root',
                          scales.PmfStat('poll'),
                          )

authPlain = cloudId + ':' + installId
auth = binascii.b2a_base64(authPlain.encode('ascii')).strip(b'=\n')

class Poller(object):
    def __init__(self, influx, graph):
        self.influx = influx
        self.graph = graph
        reactor.callLater(0, self.poll)

    @STATS.poll.time()
    @inlineCallbacks
    def poll(self):
        ret = None
        startTime = time.time()
        try:
            url = (f'http://{deviceIp}/cgi-bin/cgi_manager').encode('ascii')
            resp = yield fetch(
                url,
                method=b'POST',
                headers={b'Authorization': [b'Basic %s' % auth]},
                postdata=(f'''<LocalCommand>
                              <Name>get_usage_data</Name>
                              <MacId>0x{macId}</MacId>
                            </LocalCommand>
                            <LocalCommand>
                              <Name>get_price_blocks</Name>
                              <MacId>0x{macId}</MacId>
                            </LocalCommand>''').encode('ascii'),
                timeout=10)
            ret = json.loads(resp.body)
            log.debug(ret)
            if ret['demand_units'] != 'kW':
                raise ValueError
            if ret['summation_units'] != 'kWh':
                raise ValueError
            pts = [
                dict(measurement='housePowerW',
                     fields=dict(value=float(ret['demand']) * 1000),
                     tags=dict(house='berkeley'),
                     time=int(startTime))]
            sd = float(ret['summation_delivered'])
            if sd > 0: # Sometimes nan
                pts.append(dict(measurement='housePowerSumDeliveredKwh',
                                fields=dict(value=float()),
                                tags=dict(house='berkeley'),
                                time=int(startTime)))
            if 'price' in ret:
                pts.append(dict(
                    measurement='price',
                    fields=dict(price=float(ret['price']),
                                price_units=float(ret['price_units'])),
                    tags=dict(house='berkeley'),
                    time=int(startTime),
                ))

            self.influx.write_points(pts, time_precision='s')

            self.graph.patchObject(context=ROOM['powerEagle'],
                                         subject=ROOM['housePower'],
                                         predicate=ROOM['instantDemandWatts'],
                                         newObject=Literal(float(ret['demand']) * 1000))
        except Exception as e:
            traceback.print_exc()
            log.error("failed: %r", e)
            log.error(repr(ret))
            os.abort()

        now = time.time()
        goal = startTime + periodSec - .2
        reactor.callLater(max(1, goal - now), self.poll)


if __name__ == '__main__':
    arg = docopt("""
    Usage: reader.py [options]

    -v           Verbose
    --port PORT  Serve on port [default: 10016].
    """)
    verboseLogging(arg['-v'])

    influx = InfluxDBClient('bang', 9060, 'root', 'root', 'main')
    masterGraph = PatchableGraph()
    p = Poller(influx, masterGraph)

    reactor.listenTCP(
        int(arg['--port']),
        cyclone.web.Application(
            [
                (r'/stats/(.*)', StatsHandler, {'serverName': 'powerEagle'}),
                (r"/graph/power", CycloneGraphHandler, {'masterGraph': masterGraph}),
                (r"/graph/power/events", CycloneGraphEventsHandler, {'masterGraph': masterGraph}),
            ],
        ))
    reactor.run()