8.15. flashdb 模块 API 文档

8.15.1. API

8.15.1.1. class KVDB(_flashdb.KVDB):

def get_blob(self,key,size):...

def set_by_fmt(self,key,v,fmt):...

def get_by_fmt(self,key,size,fmt):...

8.15.1.2. class TSDB(_flashdb.TSDB):

def __init__(self,name:str,path:str,max_len:int=1024,user_data=None):...

def tsl_iter_by_time(self,from_time,to_time,callback:any,user_data:any)->int:...

8.15.1.3. class TSL(_flashdb.TSL):

8.15.2. Examples

8.15.2.1. flashdb_kvdb1.py

import flashdb
import struct

DB_PATH = "test/out/fdb_kvdb"

print('test boot_count increment 1')
boot_count = 0
boot_count_blob = struct.pack('i', boot_count)
boot_times = [0, 1, 2, 3, 0, 0, 0, 0, 0, 0]
boot_time_tuple = tuple(boot_times)
boot_time_blob = struct.pack('@10Q', *boot_time_tuple)
default_kv = {
    'username': 'armink',  # string KV
    'password': "123456",   # string KV
    'boot_count': boot_count_blob,       # int type kv
    'boot_time': boot_time_blob,   # array type kv
}
# print(default_kv)
fdb = flashdb.KVDB("env", DB_PATH, default_kv, None)

res = fdb.get_blob("boot_count", len(boot_count_blob))
assert res is not None

boot_count = struct.unpack("i", res)[0]
boot_count = boot_count+1
boot_count_blob = struct.pack('i', boot_count)

fdb.set_blob("boot_count", boot_count_blob)
res = fdb.get_blob("boot_count", len(boot_count_blob))
assert res is not None

new_boot_count = struct.unpack("i", res)[0]
assert new_boot_count == boot_count
print('PASS')

8.15.2.2. flashdb_kvdb2.py

import flashdb
import struct

DB_PATH = "test/out/fdb_kvdb"

print('test boot_count increment 2')
boot_count=0
boot_count_fmt='i'
boot_count_blob = struct.pack(boot_count_fmt, boot_count)
boot_count_size = len(boot_count_blob)

boot_time = [0,1,2,3,0,0,0,0,0,0]
boot_time_fmt='@10Q'
boot_time_tuple = tuple(boot_time)
boot_time_blob  = struct.pack(boot_time_fmt, *boot_time_tuple)

default_kv={
    'username': 'armink',  # string KV
    'password': "123456",   # string KV
    'boot_count': boot_count_blob,       # int type kv
    'boot_time': boot_time_blob,   # array type kv
}

fdb = flashdb.KVDB("env", DB_PATH, default_kv, None)
#print(default_kv)
kvdb = fdb

boot_count = fdb.get_by_fmt("boot_count", boot_count_size, boot_count_fmt)
assert boot_count is not None
print("==================== kvdb_basic_sample ====================")
print( "get the 'boot_count' value is %d" % boot_count)
        
boot_count = boot_count +1
res =fdb.set_by_fmt("boot_count", boot_count, boot_count_fmt)

new_boot_count = fdb.get_by_fmt("boot_count", boot_count_size, boot_count_fmt)
assert new_boot_count is not None
print( "get the 'boot_count' value is %d" % new_boot_count)
print("===========================================================")
assert new_boot_count == boot_count
print('PASS')

8.15.2.3. flashdb_tsdb1.py

import flashdb
import struct
import time
import os
DB_PATH = "test/out/fdb_tsdb"


tsdb = flashdb.TSDB("env", DB_PATH, max_len=512)

tic = time.time() * 1000
for i in range(10):
    blob_i = struct.pack('i', i)
    time.sleep(0.001)
    ret = tsdb.tsl_append(blob_i)

toc = time.time() * 1000
assert ret == 0


sum_by_time = 0


def callback(tsl, user_data) -> int:
    global sum_by_time
    # print(tsl.get_time(), tsl.to_blob())
    t = tsl.get_time()
    blob_i = tsl.to_blob()
    i = struct.unpack('i', blob_i)[0]
    print(t, i, user_data)
    if user_data == 'user_data_by_time':
        sum_by_time += i
    return False  # False: continue, True: stop

assert tsdb.tsl_iter(callback, 'user_data') == 0
assert tsdb.tsl_iter_reverse(callback, 'user_data_reverse') == 0
print('toc - tic', toc - tic)
assert tsdb.tsl_iter_by_time(tic, toc, callback, 'user_data_by_time') == 0
assert sum_by_time == 45

print('PASS')