Python3 sqlite3从json数据加载表

CREATE TABLE DATA ( time_index INTEGER, TimeStamp INTEGER, EUI64 TEXT, modelident TEXT, modelname TEXT, alias TEXT, -- added this - check R script for compatability devicetype INTEGER, voltage REAL, current REAL, frequency REAL, powerfactory REAL, activepower REAL, apparentpower REAL, mainenergy REAL, voltage2 REAL, current2 REAL, frequency2 REAL, powerfactory2 REAL, activepower2 REAL, apparentpower2 REAL, mainenergy2 REAL, voltage3 REAL, current3 REAL, frequency3 REAL, powerfactory3 REAL, activepower3 REAL, apparentpower3 REAL, mainenergy3 REAL, voltage4 REAL, current4 REAL, frequency4 REAL, powerfactory4 REAL, activepower4 REAL, apparentpower4 REAL, mainenergy4 REAL, Dimming_L1 REAL, Dimming_L2 REAL, Dimming_L3 REAL, Dimming_L4 REAL, Temperature REAL, Humidity REAL, onoffstatus INTEGER, rssi INTEGER, lqi INTEGER, nenergy REAL, rssi1 INTEGER, CT_ratio INTEGER ); CREATE TABLE index_Table ( num INTEGER, HS_index INTEGER, UpDateIndex INTEGER ); Device ID Usage 000D6F0005A5D77E Power Points 000D6F0005A5BCAE Shed 000D6F0005A5BE9D Grid So, the cmd=list_metering generates json output similar to the following { "devices": [ { "signalstrength1": "N/A", "negativeenergy": "0.072", "mainenergy": "11282.062", "apparentpower": "805.22", "activepower": "698.76", "powerfactor": "87", "frequency": "50.02", "current": "3.38", "voltage": "238.43", "timestamp": "1609152925", "signalstrength": "40", "dimming": "false", "sense": "false", "supportIdentify": "1", "state": "N/A", "metering3phase": "false", "metering": "true", "alias": "Power Points", "modid": "gpm-hazr-4.2", "model": "SG3015-T3(100A)", "deviceid": "000D6F0005A5D77E" }, { "signalstrength1": "N/A", "negativeenergy": "8272.385", "mainenergy": "11695.240", "apparentpower": "562.74", "activepower": "-32.76", "powerfactor": "-6", "frequency": "50.02", "current": "2.36", "voltage": "238.32", "timestamp": "1609152925", "signalstrength": "44", "dimming": "false", "sense": "false", "supportIdentify": "1", "state": "N/A", "metering3phase": "false", "metering": "true", "alias": "Grid", "modid": "gpm-hazr-4.2", "model": "SG3015-T3(100A)", "deviceid": "000D6F0005A5BE9D" }, { "signalstrength1": "N/A", "negativeenergy": "50.763", "mainenergy": "11534.174", "apparentpower": "228.53", "activepower": "112.86", "powerfactor": "49", "frequency": "50.02", "current": "0.96", "voltage": "238.05", "timestamp": "1609152925", "signalstrength": "40", "dimming": "false", "sense": "false", "supportIdentify": "1", "state": "N/A", "metering3phase": "false", "metering": "true", "alias": "Shed", "modid": "gpm-hazr-4.2", "model": "SG3015-T3(100A)", "deviceid": "000D6F0005A5BCAE" } ], "status": "success", "cmd": "list_metering", "version": "1.1" }

1条回答

网友

1楼 · 发布于 2024-09-28 01:33:22

我不是python专家，所以对于python3编码风格的改进建议非常感谢

下面的代码工作正常，不整洁，没有任何合理的错误处理，但它是概念证明

    with requests.Session() as session:
        #print("Retrieving readings from Power Gateway...")
        cmd_list_metering_json_result = session.get("http://" + host + "/api_json.asp?cmd=list_metering&auth="+userPasswordBase64)
        #print("Retrieved.")
        cmd_list_metering_dict = cmd_list_metering_json_result.json()
        #print("Number of results: " , len(cmd_list_metering_dict))
        #print("cmd:list_metering:returned:", cmd_list_metering_dict)
        #print(json.dumps(cmd_list_metering_dict, indent=4, sort_keys=True))

        #print("All keys ", cmd_list_metering_dict.keys())

        if cmd_list_metering_dict['status'] != 'success':
            syslog.syslog("FAILURE to retrieve cmd_list_metering from power gateway")
            exit(1)
        #
        ## Only interested in the 'devices' key
        ## which is a list of dictionaries, one dictionary per device
        ## 
        #
        #print("All Devices = ",  cmd_list_metering_dict['devices'])
        #print("Type = ", type( cmd_list_metering_dict['devices']) )

        numberOfDevices = len(cmd_list_metering_dict['devices'])
        #print("Number of devices = ", numberOfDevices)
        if numberOfDevices != 3:
            syslog.syslog("PowerTracker requires reboot - expect 3 devices")
            exit(1)
                    
        devIx = 0
        timeOfDayInSeconds = int(datetime.datetime.today().timestamp())
        #print("timeOfDayInSeconds = ", timeOfDayInSeconds)
        for devIx in range(0, numberOfDevices):
            device_dict = cmd_list_metering_dict['devices'][devIx]
            #print("Storing device ", device_dict['deviceid'], " alias ", device_dict['alias'])
            data_tuple = (
                           timeOfDayInSeconds   
                          ,device_dict['timestamp']
                          ,device_dict['deviceid']
                          ,device_dict['modid']
                          ,device_dict['model']
                          ,device_dict['alias']
                          ,15                    
                          ,device_dict['voltage']
                          ,device_dict['current']
                          ,device_dict['frequency']
                          ,device_dict['powerfactor']
                          ,device_dict['activepower']
                          ,device_dict['apparentpower']
                          ,device_dict['mainenergy']
                          ,1
                          ,device_dict['signalstrength']
                          ,1
                          ,device_dict['negativeenergy']
                          ,1
                          )
            
            rowsInserted = dbCursor.execute("""insert into DATA (
   time_index
  ,TimeStamp      
  ,EUI64          
  ,modelident     
  ,modelname      
  ,alias          
  ,devicetype     
  ,voltage        
  ,current        
  ,frequency      
  ,powerfactory   
  ,activepower    
  ,apparentpower  
  ,mainenergy     
  ,onoffstatus    
  ,rssi
  ,lqi           
  ,nenergy        
  ,CT_ratio
) values (
   ?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
  ,?
    );""", data_tuple)

    #print("Committing transaction...")
    dbConn.commit()
    #print("Closing cursor...")
    dbCursor.close()            
    #if dbConn:
    #    print("Disconnecting from database...")
    #    dbConn.close()

相关问题更多 >

编程相关推荐

热门问题

热门文章