NewProducer



In [8]:
cd git/SolarDataRESTfulAPI/


[Errno 2] No such file or directory: 'git/SolarDataRESTfulAPI/'
/home/iot/repos/SolarDataRESTfulAPI

In [9]:
import json
import pandas as pd
import InfluxDBInterface
import time
reload(InfluxDBInterface)
from ElasticsearchInterface import ESinterface
import IoTtoolkit
#reload(IoTtoolkit)
import sys
import mosquitto
import os
import argparse

DataLink = InfluxDBInterface.InfluxDBInterface("influxInterfaceCredentials2.json")

LogDB = DataLink.databases[u'SolarLogdata']
ProductionDB = DataLink.databases[u'SolarProductionSites']
AreaDB = DataLink.databases[u'SolarProductionAreas']
Test = DataLink.databases[u'test']

es = ESinterface()

In [40]:
def AreaProductionAlgorithm(SitesProductionBuf,sites,PowerStreams,EnergyStreams):
    dfPower = SitesProductionBuf.Data[PowerStreams]
    dfEnergy = SitesProductionBuf.Data[EnergyStreams]
    AreaProduction = pd.DataFrame()
    
    
    AreaProduction["Power"] = dfPower.sum(1)
    AreaProduction["nSumedPower"] = dfPower.count(1)
    #AreaProduction["PmaxSumed"]
    
    Pmax = pd.DataFrame(list(sites.loc["Pmax"]),index=list(sites.loc["online_since"]/1000),columns = ["Pmax"])#dfPower #AreaProduction
    Pmax.sort_index(inplace=True)
    Pmax["Pmax"] *= 1000
    Pmax["nMax"] =1

    #Save the period.
    Start = AreaProduction.index[0]
    Stop = AreaProduction.index[-1]
    
    #Add Pmax
    AreaProduction = AreaProduction.join(Pmax.cumsum(),how="outer")
    
    #Calculate PmaxSum
    df = pd.DataFrame(sites.loc["Pmax"]).T
    df = df.reindex_axis(sorted(df.columns), axis=1)
    df2 = dfPower.reindex_axis(sorted(dfPower.columns), axis=1)
    
    AreaProduction["PmaxSum"] = pd.DataFrame(df2.notnull().values * df.values, index = df2.index, columns = df.columns).sum(1)*1000
    
    #Add Energy and nSummedEnergy 
    AreaProduction = AreaProduction.join(pd.DataFrame(dfEnergy.ffill().sum(1),columns = ["Energy"]),how="outer")
    AreaProduction["nSumedEnergy"] = dfEnergy.count(1)
    
    
    #Cut out the part we need.
    AreaProduction.fillna(method="ffill", inplace = True)
    AreaProduction = AreaProduction.loc[Start:Stop]
    
    return AreaProduction

In [63]:
def CalculateAreaProduction(area,Recalculate=False):
    global AreaProduction2
    global debugName
    
    sites = es.GetHitsMatchingPropDict("solar-sites-index","meta-data",json.loads(area["query"]))
    
    if sites.shape[1] == 0:
        return 
    
    SitesProduction = IoTtoolkit.Feed()
    
    PowerStreams = SitesProduction.CombineStreamsFromMulipleSources("Power",ProductionDB,sites.columns,"Power",Compressed=False)
    EnergyStreams = SitesProduction.CombineStreamsFromMulipleSources("Energy",ProductionDB,sites.columns,"Energy",Compressed=True)
    SitesProductionBuf = SitesProduction.GetBuffer()
    SitesProductionBuf.Size = 5000
    
    if Recalculate:
        SitesProductionBuf.Seek(0)
    else:
        #Recalculate last week and forward. 
        SitesProductionBuf.Seek(time.time() - (14 * 24 * 3600))
    
    
    while not SitesProductionBuf.EOF:
        AreaProduction = AreaProductionAlgorithm(SitesProductionBuf,sites,PowerStreams,EnergyStreams)
        try:
            AreaDB.Replace(area.name,AreaProduction,Compressed = False)
        except ValueError:
            print AreaProduction
            AreaProduction2 = AreaProduction
            debugName = area.name
            break
        SitesProductionBuf.Next()

In [81]:
if True: #if __name__ == "__main__":
    
    #Parse arguments
    #parser = argparse.ArgumentParser(add_help=False)
    #parser.add_argument('-h', dest='host', default="localhost", help='MQTT host send results to')
    #parser.add_argument('-t', dest='topic', default="", help='MQTT topic to process')
    #parser.add_argument('-m', dest='message', default="", help='MQTT message to process')

    #args = parser.parse_args()

    #Get location of script
    path = os.path.abspath(os.path.dirname(sys.argv[0]))


    #Set up MQTT
    ip = "localhost"
    port = 1883
    user = "driver"
    password = "1234"
    prefix = "SolarAreaProductionProducer"
    
    mqtt=mosquitto.Mosquitto("AreaProductionProducer")
    mqtt.prefix = prefix
    mqtt.ip = ip
    mqtt.port = port
    #mqtt.clientId = clientId
    mqtt.user = user
    mqtt.password = password
                
    if mqtt != None:
        mqtt.username_pw_set(user,password)
    
    #mqtt.will_set( topic =  "system/" + prefix, payload="Idle", qos=1, retain=True)
    mqtt.connect(ip,keepalive=10)
    mqtt.publish(topic = "system/"+ prefix, payload="Updating", qos=1, retain=True)
    
    areas = es.GetHitsAsDataFrame("solar-area-index","meta-data")
    
    #areas = areas.iloc[:,-7:]
    
    for id in areas.columns: 
        print "Processing %s" % areas[id]["Name"]
        print "..."
        time.sleep(0.5)
        CalculateAreaProduction(areas[id],True)
    
    print "All done!"
    
    mqtt.connect(ip,keepalive=10)
    #mqtt.publish(topic = "solardata/production/at", payload=str((TrailTime,LeadTime)), qos=1, retain=True) 
    mqtt.publish(topic = "solardata/area-production/lastupdate", payload=now, qos=1, retain=True)    
    mqtt.publish(topic = "system/"+ prefix, payload="Idle", qos=1, retain=True)


Got 19 Hits:
Processing Norrmalm
...
Got 0 Hits:
Processing Bromma
...
Got 0 Hits:
Processing Skarpnäck
...
Got 0 Hits:
Processing Husby
...
Got 11 Hits:
Processing Rinkeby
...
Got 7 Hits:
Processing Farsta
...
Got 0 Hits:
Processing Spånga-Tensta
...
Got 0 Hits:
Processing Skärholmen
...
Got 0 Hits:
Processing Akalla
...
Got 0 Hits:
Processing Gärdet
...
Got 1 Hits:
Processing Södermalm
...
Got 0 Hits:
Processing Enskede-Årsta-Vantör
...
Got 0 Hits:
Processing Rinkeby-Kista
...
Got 18 Hits:
Processing Kista 
...
Got 0 Hits:
Processing Hägersten-Liljeholmen
...
Got 0 Hits:
Processing Hässelby-Vällingby
...
Got 0 Hits:
Processing Östermalm
...
Got 1 Hits:
Processing Kungsholmen
...
Got 0 Hits:
Processing Älvsjö
...
Got 0 Hits:
---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
<ipython-input-81-07b0a20e7912> in <module>()
     49     mqtt.connect(ip,keepalive=10)
     50     #mqtt.publish(topic = "solardata/production/at", payload=str((TrailTime,LeadTime)), qos=1, retain=True)
---> 51     mqtt.publish(topic = "solardata/area-production/lastupdate", payload=now, qos=1, retain=True)
     52     mqtt.publish(topic = "system/"+ prefix, payload="Idle", qos=1, retain=True)
     53 

NameError: name 'now' is not defined
All done!

In [59]:
AreaProduction2


Out[59]:
Power nSumedPower Pmax nMax PmaxSum Energy nSumedEnergy
1396930200 1588 10 229300 9 270500 1166300 10
1396930200 1588 10 244600 10 270500 1166300 10
1396932600 1 1 244600 10 28000 1166300 10
1396933200 9 1 244600 10 28000 1166300 10
1396933800 51 2 244600 10 63500 1166300 10
1396934400 408 4 244600 10 102600 1166300 10
1396935000 1136 5 244600 10 121700 1166300 10
1396935600 1426 5 244600 10 121700 1166000 10
1396936200 1199 5 244600 10 121700 1166000 10
1396936800 1444 5 244600 10 121700 1166100 10
1396937400 1435 4 244600 10 102700 1166300 10

In [52]:
print "Hej hå"


Hej hå

In [55]:
time.time()


Out[55]:
1405197126.862326

In [57]:
hits = es.GetHitsMatchingPropDict("solar-sites-index","meta-data",{"Operator":"SLB"})


Got 0 Hits:

In [17]:
site = "2a31fb24-347b-4924-ab89-5c434771a2ae"
timestamp = time.time()

In [21]:
test = ProductionDB.GetDataAfterTime(site,timestamp=timestamp-60*60*5,limit = 100)

In [86]:
def EpocToDate(timestamp):
    return time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(timestamp))

Series = ProductionDB.ListSeries()
for serie in Series:
 print serie, EpocToDate(ProductionDB.GetFirstTimestamp(serie)/1000)


 2a31fb24-347b-4924-ab89-5c434771a2ae 2014-03-05 17:00:00
32383bde-7648-4abe-9dac-44701eabd72f 2014-04-09 06:10:00
46d55815-f927-459f-a8e2-8bbcd88008ee 2013-10-16 07:20:00
4a39b124-f594-11e3-a510-f23c9173ce4a 2014-06-06 03:30:00
4cf6c743-8d0b-45f0-aa72-8a0c10315cf5 2014-03-26 05:30:00
6f075b69-b823-4e6b-8021-dd751cc79eea 2014-03-06 08:50:00
709e47a1-ca88-4c22-890f-2407e9cb131a 2014-04-08 06:10:00
71b0e5ec-1da8-11e4-a510-f23c9173ce4a 2014-06-09 17:20:00
79ea7d0e-1da8-11e4-a510-f23c9173ce4a 2014-06-09 03:20:00
802afb51-c5eb-4230-9b22-9a77ef7260b3 2014-03-05 16:30:00
845e5e54-1da8-11e4-a510-f23c9173ce4a 2014-06-11 17:30:00
8b28b202-1da8-11e4-a510-f23c9173ce4a 2014-06-11 19:20:00
916b6e8e-1da8-11e4-a510-f23c9173ce4a 2014-06-12 09:10:00
b0dd71a8-efe8-41eb-bab1-633ebaaa778c 2014-04-10 10:30:00
c6261166-f651-11e3-a510-f23c9173ce4a 2014-04-07 06:00:00
e63710a4-78d9-4071-9a2e-ad35534355f4 2014-03-05 12:30:00
e8a13982-f651-11e3-a510-f23c9173ce4a 2014-04-07 06:00:00
f1136e33-d227-4fbd-ae72-d3488d70ba89 2014-03-06 08:10:00
face0d13-b22d-48fe-bec1-d29732604133 2014-04-08 06:10:00

In [39]:
test.iloc[-1].Power


Out[39]:
0.0

In [44]:
test["Energy"].max()


Out[44]:
18176800.0

In [4]:
cd /home/iot/repos/SolarDataRESTfulAPI/


/home/iot/repos/SolarDataRESTfulAPI

In [5]:
ls


cvs_importer.py             influxInterfaceCredentials2.json  SLB_DataImporter.py*
ElasticsearchInterface.py   IoTtoolkit.py                     SolarAreaProductionProducer.py*
ElasticsearchInterface.pyc  IoTtoolkit.pyc                    SolarDataRESTapi.py
Feeds.json                  LICENSE                           SolarProductionProducer.py*
InfluxDBInterface.py        README.md                         SolarRestAPI.sh*
InfluxDBInterface.pyc       SeriesStatus.py*                  Universe.json
influxDBSolarREST.py        SiteIDs.json

In [6]:
cat influxInterfaceCredentials2.json


[{"host":"livinglab2.powerprojects.se",
 "port":8086,
 "user":"uploader",
 "password":"ryKkSSnveKVpUMROt8kqvZCGJXJveu8MkJO",
 "database":"SolarLogdata"},
{"host":"livinglab2.powerprojects.se",
 "port":8086,
 "user":"uploader",
 "password":"ryKkSSnveKVpUMROt8kqvZCGJXJveu8MkJO",
 "database":"SolarProductionSites"},
{"host":"livinglab2.powerprojects.se",
 "port":8086,
 "user":"uploader",
 "password":"ryKkSSnveKVpUMROt8kqvZCGJXJveu8MkJO",
 "database":"SolarProductionAreas"},
{"host":"livinglab2.powerprojects.se",
 "port":8086,
 "user":"uploader",
 "password":"ryKkSSnveKVpUMROt8kqvZCGJXJveu8MkJO",
 "database":"test"}]



In [ ]: