xumu_iotdb/config.py

# 数据库插入类型
import base64

rfid_type = ["TEXT", "TEXT", "FLOAT", "INT32"]
air_type = ["TEXT", "FLOAT", "FLOAT", "INT32"]
else_type = ["TEXT", "FLOAT", "INT32"]
dataTypes = {
    0: rfid_type,
    1: air_type,
    2: else_type,
    3: else_type,
    4: else_type,
    5: else_type,
}
baseHost = "https://iot.lihaink.cn/iotdb_restapi"
# 注意这里前面不能加/
insertUri = "rest/v2/insertRecords"
queryUri = "rest/v2/query"
# 鉴权
username = 'root'
password = 'root'
code = (username + ":" + password).encode("utf-8")
token = base64.encodebytes(code).decode("utf-8").strip()
headers = {
    'ContentType': 'application/json',
    'Authorization': "Basic " + token
}
# RFID
send1 = {
    "m": ["cid", "r", "v", "t"],
    "v": ["abcd", "rfid", 10.62, 0],
    "l": 63
}
# 空气
send2 = {
    "m": ["cid", "at", "ah", "t"],
    "v": ["test2", 10.62, 50.22, 1],
    "l": 65
}
# 普通
send3 = {
    "m": ["cid", "v", "t"],
    "v": ["test1", 10.62, 2],
    "l": 50
}


def RFID_template(deviceId):
    return [
        f"CREATE TIMESERIES root.farm.{deviceId}.iccid(cid) WITH datatype=TEXT,ENCODING=PLAIN",
        f"CREATE TIMESERIES root.farm.{deviceId}.temperature(v) WITH datatype=FLOAT,ENCODING=PLAIN",
        f"CREATE TIMESERIES root.farm.{deviceId}.RFID(r) WITH datatype=TEXT,ENCODING=PLAIN",
        f"CREATE TIMESERIES root.farm.{deviceId}.type(t) WITH datatype=INT32,ENCODING=PLAIN",
    ]


def common_template(deviceId):
    return [
        f"CREATE TIMESERIES root.farm.{deviceId}.iccid(cid) WITH datatype=TEXT,ENCODING=PLAIN",
        f"CREATE TIMESERIES root.farm.{deviceId}.value(v) WITH datatype=FLOAT,ENCODING=PLAIN",
        f"CREATE TIMESERIES root.farm.{deviceId}.type(t) WITH datatype=INT32,ENCODING=PLAIN",
    ]


def air_template(deviceId):
    return [
        f"CREATE TIMESERIES root.farm.{deviceId}.iccid(cid) WITH datatype=TEXT,ENCODING=PLAIN",
        f"CREATE TIMESERIES root.farm.{deviceId}.air_temperature(at) WITH datatype=FLOAT,ENCODING=PLAIN",
        f"CREATE TIMESERIES root.farm.{deviceId}.air_humidity(ah) WITH datatype=FLOAT,ENCODING=PLAIN",
        f"CREATE TIMESERIES root.farm.{deviceId}.type(t) WITH datatype=INT32,ENCODING=PLAIN",
    ]


# 数据库创建字段sql语句
def get_sql(deviceId, type):
    match type:
        case 0:
            return RFID_template(deviceId)
        case 1:
            return air_template(deviceId)
        case _:
            return common_template(deviceId)