1

由于 COSM 已成为 Xively，因此添加了一个不错的设备 api（或者一直不确定）。流量是

  1. 使用序列号创建产品批次
  2. 使用某些产品批次标识符 (?) 激活设备
  3. 使用获得的 feed/api 密钥开始使用设备

我不知道如何通过 python API 做到这一点-有任何指针吗？

4

2 回答 2

0

这应该添加到库中，但现在您可以使用此代码来实现设备激活。我已经使用环境变量来存储产品密钥和设备序列号，但可以根据您的用例进行更改。唯一棘手的部分是您需要调用a2b_hex().

import xively

from os import environ    
from hashlib import sha1
from binascii import a2b_hex
import hmac

secret = environ['XIVELY_PRODUCT_SECRET']
serial = environ['XIVELY_DEVICE_SERIAL_NUMBER']

activation = hmac.new(a2b_hex(secret), serial, sha1).hexdigest()

creds = xively.Client(key=None).get('/v2/devices/'+activation+'/activate').json()

xi_feed = xively.XivelyAPIClient(creds['apikey']).feeds.get(creds['feed_id'])

您还需要注意将凭据存储到文件中，因为设备只能激活一次。如果您尝试一次又一次地运行此代码，您会注意到 403 错误，因此请使用 Xively 开发人员工作台来停用被测设备（您可能需要刷新页面）。

这是一个使用配置文件或环境变量的完整示例：

#!/usr/bin/python

from os import environ 
from hashlib import sha1 
from binascii import a2b_hex 
import hmac
import sys, subprocess
import ConfigParser
import xively

CONFIG_FILE = 'xively.conf'

PROVISIONING = 'PROVISIONING'

PROVISIONING_PRODUCT_SECRET = 'PRODUCT_SECRET'
PROVISIONING_DEVICE_SERIAL = 'DEVICE_SERIAL'

PROVISIONING_FEED_ID = 'FEED_ID'
PROVISIONING_API_KEY = 'API_KEY'

def get_setting(config, section, key):
    try:
    value = config.get(section, key)
    except:
    print key + " not found in config file. Using environment variable " + key + " instead."

    try:
        value = environ[key]
    except:
        print key + " not found in environment."
        raise
    # value defined?
    if not value:
    raise
    return value

def xively_activate_product(secret, serial):
    activation = hmac.new(a2b_hex(secret), serial, sha1).hexdigest()
    creds = xively.Client(key=None).get('/v2/devices/'+activation+'/activate').json()
    return creds 

# main
config = ConfigParser.RawConfigParser()
config.read(CONFIG_FILE)

try:
    # see if we already have an api key and feed id
    feed_id = config.get(PROVISIONING, PROVISIONING_FEED_ID)
    api_key = config.get(PROVISIONING, PROVISIONING_API_KEY)

    print "Provisioned product details:"
    print "FEED_ID: " + str(feed_id)
    print "API_KEY: " + api_key

    # continue working with your activated product here

except:
    print "FEED_ID and API_KEY not found. Activating product now."

    # no error handling for secret- it _is_ needed
    try:
    secret = get_setting(config, PROVISIONING, PROVISIONING_PRODUCT_SECRET)
    except:        
    print "Finding " + PROVISIONING_PRODUCT_SECRET + " failed. Giving up."
    sys.exit(1)

    try:
    serial = get_setting(config, PROVISIONING, PROVISIONING_DEVICE_SERIAL)
    except:        
    serial = subprocess.check_output("hostname", shell=True)
    if not serial:
        print "Fallback to hostname for " + PROVISIONING_DEVICE_SERIAL + " failed. Giving up."
        sys.exit(1)

    try:    
    creds = xively_activate_product(secret, serial)

    # check if there were errors
    try:
        creds["errors"]
    except:
        pass
    else:
        print "Product activation failed (" + creds["title"] +": "+ creds["errors"] + ")."
        sys.exit(1)

    feed_id = creds['feed_id']
    api_key = creds['apikey']

    print "Product activation successful."
    print "FEED_ID: " + str(feed_id)
    print "API_KEY: " + api_key

    if not config.has_section(PROVISIONING):
        config.add_section(PROVISIONING)

    config.set(PROVISIONING, PROVISIONING_FEED_ID, feed_id)
    config.set(PROVISIONING, PROVISIONING_API_KEY, api_key)

    # Writing our configuration file to 'example.cfg'
    with open(CONFIG_FILE, 'wb') as configfile:
        config.write(configfile)

    except Exception as e:
    print "Product activation failed (" + str(e) +")."
    sys.exit(1)
于 2013-06-24T10:02:57.977 回答
0

这是我写的另一个有用的课程：

## Logging for debugging purposes
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


import os
from os import environ 
from hashlib import sha1 
from binascii import a2b_hex 
import hmac
import sys, subprocess
import ConfigParser
import xively


PROVISIONING = 'PROVISIONING'

PROVISIONING_PRODUCT_SECRET = 'PRODUCT_SECRET'

PROVISIONING_FEED_ID = 'FEED_ID'
PROVISIONING_API_KEY = 'API_KEY'


class XivelyManager:

    def __init__(self, settings="xively.conf"):
        # main
        self.settings=settings
        self.config = ConfigParser.RawConfigParser()
        self.config.read(settings)

        try:
            # see if we already have an api key and feed id

            self.api_key = self.get_setting(PROVISIONING, PROVISIONING_API_KEY)
            self.secret = self.get_setting(PROVISIONING, PROVISIONING_PRODUCT_SECRET)
            # continue working with your activated product here

        except:
            logger.exception( "API KEY and SECRET NOT FOUND" )

    def activate_sensor(self,serial):

            try:    
                creds = self.xively_activate_product(str(serial))

                # check if there were errors
                try:
                    creds["errors"]
                except:
                    pass
                else:
                    logger.exception("Product activation failed (" + creds["title"] +": "+ creds["errors"] + ").")
                    return False

                feed_id = creds['feed_id']
                api_key = creds['apikey']

                if not self.config.has_section(PROVISIONING):
                    self.config.add_section(PROVISIONING)

                if not self.config.has_section(str(serial)):
                    self.config.add_section(str(serial))

                self.config.set(PROVISIONING, PROVISIONING_API_KEY, api_key)
                self.config.set(str(serial), PROVISIONING_FEED_ID , feed_id)

                # Writing our configuration file to 'xively.cfg'
                with open(self.settings, 'wb') as configfile:
                    self.config.write(configfile)
                return True

            except Exception as e:
                logger.exception("Product activation failed (" + str(e) +").")
                return False

    def get_setting(self, section, key):
        try:

            value = self.config.get(section, key)
        except:
            logger.exception( key + " not found in config file. Using environment variable " + key + " instead.")
##            try:
##                value = environ[key]
##            except:
##                logger.exception( key + " not found in environment.")
##            finally:
##                pass
        finally:

            # value defined?
            if not value:
                raise
            return value

    def get_feed(self,serial):
        try:
            if self.config.has_section(str(serial)):
                feed_id = self.get_setting(str(serial), PROVISIONING_FEED_ID)
            else:
                feed_id=False
        except Exception, e:
            feed_id=False
        finally:
            return feed_id

    def xively_activate_product(self, serial):
        activation = hmac.new(a2b_hex(self.secret), serial, sha1).hexdigest()
        creds = xively.Client(key=None).get('/v2/devices/'+activation+'/activate').json()
        return creds 

if __name__ == "__main__":
    print "Testing Xively Manager "
    settings = os.path.join(os.path.dirname(sys.argv[0]), "config", "xively.conf")
    print settings

    testxive=XivelyManager(settings)
    #print testxive.activate_sensor(10)

    print testxive.get_feed(10)

当您的互联网网关连接到其他几个设备时，这很有用。您的配置文件将使用以下内容进行更新：

[PROVISIONING]
product_secret = xxxxxxxxxxxxxxxxxxxxxxxxxxxx
api_key = xxxxxxxxxxxxxxxxxxxxxxxx

[productserial1]
feed_id = xxxxxxxx

[productserial2]
feed_id = xxxxxxxx
于 2013-08-28T09:30:52.577 回答