How to Save Movella DOT Data to a SQLite Database
Store your human movement data easily with SQLite and Python on Mac OS.
You’ve completed our other tutorials on How to Connect to or Stream Data from Movella DOT wearable sensors with a Mac and Python. But what do we do with all that data? How do we store it?
Let’s use SQLite, a lightweight database, to store it and easily access it, so that we can do data analysis and visualization on it later.
What is SQLite?
SQLite is the toy version of big SQL databases like MySQL and Postgres, which makes it ideal for small projects.
Whereas bigger databases require passwords and database servers, SQLite allows us to use the power of the SQL language on what is basically a fancy text file. No passwords, no servers, no ports, etc.
Unlike text files or CSVs, SQLite will allow us to:
- grab data out of order
- use the data in a live dashboard or other real-time visualization
Bonus: SQLite is built-in to every Python install.
End Goal
Save data from two Movella DOTs to a SQLite database, so that we can query it, analyze it, or use it in a live dashboard.
We’ll do this with Python 3 on a Mac computer, along with the Bleak, NumPy and SQLite libraries.
Requirements
- a Mac computer (Windows/Linux might work with some tweaking)
- Python 3
- Bleak (a Python Bluetooth LE library)
- NumPy (a Python math library)
- 1+ Movella DOT sensors
Steps
- Copy the code from our Movella DOT streaming tutorial
- Create a SQLite database
- Create a table with columns
- Format the DOT data
- Write the DOT data to the database as it comes in
- Save the database
The full script is available at the end of this tutorial.
Step 1: Copy the code from our last tutorial
We assume you’ve gone through our tutorial on How to Stream Data from Multiple Movella DOTs.
We’ll start where that tutorial leaves off with the following Python script. It is available at the very bottom of the streaming tutorial.
# Stream Free Acceleration data from multiple Movella DOTs
import numpy as np
import asyncio
from bleak import BleakClient
measurement_char_uuid = "15172001-4947-11e9-8646-d663bd873d93"
short_payload_char_uuid = "15172004-4947-11e9-8646-d663bd873d93"
# Replace this with a list of your DOT's UUID addresses
# (or "MAC addresses" for Windows and Linux users)
addresses = [
"509808FF-ECFE-895D-C1FE-BE5AC5DB6204",
"338312FA-C3D1-183F-325A-0726AFDBEB78"
]
class NotificationHandler:
'''This class allows us to add the DOT's UUID address to the data that gets printed'''
def __init__(self, device_address):
self.device_address = device_address
def callback(self, sender, data):
free_acceleration = encode_free_acceleration(data)[0]
free_acceleration = str(free_acceleration)[1:-1]
print(f"{self.device_address} -- {free_acceleration}")
def encode_free_acceleration(bytes_):
data_segments = np.dtype([
('timestamp', np.uint32),
('x', np.float32),
('y', np.float32),
('z', np.float32),
('zero_padding', np.uint32)
])
formatted_data = np.frombuffer(bytes_, dtype=data_segments)
return formatted_data
async def connect(address):
nh = NotificationHandler(address)
# Connect to the DOT and stream data
async with BleakClient(address) as client:
print(f"Client connection to `{client.address}: {client.is_connected}")
# Subscribe to data notifications
await client.start_notify(short_payload_char_uuid, nh.callback)
# Set and turn on measurement mode
binary_message = b"\x01\x01\x06"
await client.write_gatt_char(measurement_char_uuid, binary_message, response=True)
# Stream data for 10 seconds
await asyncio.sleep(10.0)
async def main():
await asyncio.gather(*(connect(addr) for addr in addresses))
if __name__ == "__main__":
asyncio.run(main())Right now, this script streams Free Acceleration data (accelerometer X, Y, and Z) from multiple Movella DOTs and prints it to Terminal:
We’ll send all of this data to a SQLite database in the next steps.
Step 2: Create a SQLite Database
The easiest way to create a SQLite database in Python for any project is:
import sqlite3
conn = sqlite3.connect("test.db")SQLite will connect to a database called test.db. If one doesn’t exist, SQLite will create one.
Multiple Databases
What we’re going to do is actually create multiple databases, one for each device. This is unorthodox, but needed, because SQLite won’t let multiple asyncio processes write to one database at the same time. However, we don’t want to upgrade to a bigger SQL version, because it requires servers and user accounts.
The next steps after that are:
- Create a table to hold our data (like a spreadsheet)
- Add rows to the table with DOT measurement data
We’ll end up with multiple movella_dot*.db files, with * being some number to differentiate them by DOT device. For example: movella_dot1.db and movella_dot2.db.
Creating the DOT databases
To create our databases, we’ll add this code to stream.py:
import sqlite3
...
def start_database(filename):
conn = sqlite3.connect(filename)
return conn
...
# This is the original connect() from the streaming script
async def connect(address):
number = address[:2]
connection = start_database(f"movella_dot{number}.db")
database = connection.cursor()
...We import SQLite, create a function to connect to the databases (start_database()), then call that function with our custom database name, movella_dot{number}.db.
To give the databases unique names, we grab the first two characters from the DOT’s address and put it in the database name with
f"...{number}". (This type of formatting is called an f-string.)
If we run the code as-is, all it will do is create a few blank databases named movella_dot*.db. Since we’re using 2 DOTs for this example, we get two databases, movella_dot50.db and movella_dot33.db.
This is the full script with database-creation code:
# Stream Free Acceleration data from multiple Movella DOTs
import asyncio
import numpy as np
import sqlite3
from bleak import BleakClient
measurement_char_uuid = "15172001-4947-11e9-8646-d663bd873d93"
short_payload_char_uuid = "15172004-4947-11e9-8646-d663bd873d93"
# Replace this with a list of your DOT's UUID addresses
# (or "MAC addresses" for Windows and Linux users)
addresses = [
"509808FF-ECFE-895D-C1FE-BE5AC5DB6204",
"338312FA-C3D1-183F-325A-0726AFDBEB78"
]
class NotificationHandler:
'''This class allows us to add the DOT's UUID address to the data that gets printed'''
def __init__(self, device_address):
self.device_address = device_address
def callback(self, sender, data):
free_acceleration = encode_free_acceleration(data)[0]
free_acceleration = str(free_acceleration)[1:-1]
print(f"{self.device_address} -- {free_acceleration}")
def start_database(filename):
conn = sqlite3.connect(filename)
return conn
def encode_free_acceleration(bytes_):
data_segments = np.dtype([
('timestamp', np.uint32),
('x', np.float32),
('y', np.float32),
('z', np.float32),
('zero_padding', np.uint32)
])
formatted_data = np.frombuffer(bytes_, dtype=data_segments)
return formatted_data
async def connect(address):
# Connect to the database (and create it if it doesn't exist yet)
number = address[:2]
connection = start_database(f"movella_dot{number}.db")
database = connection.cursor()
nh = NotificationHandler(address)
# Connect to the DOT and stream data
async with BleakClient(address) as client:
print(f"Client connection to `{client.address}: {client.is_connected}")
# Subscribe to data notifications
await client.start_notify(short_payload_char_uuid, nh.callback)
# Set and turn on measurement mode
binary_message = b"\x01\x01\x06"
await client.write_gatt_char(measurement_char_uuid, binary_message, response=True)
# Stream data for 10 seconds
await asyncio.sleep(10.0)
async def main():
await asyncio.gather(*(connect(addr) for addr in addresses))
if __name__ == "__main__":
asyncio.run(main())Let’s add a table to store the DOT data in.
Step 3: Create a table in each database
To create a table, we’ll use the SQL CREATE command inside of cursor.execute().
def create_database_table(cursor):
cursor.execute('''
CREATE TABLE IF NOT EXISTS free_acceleration (
id integer PRIMARY KEY,
bluetooth_address char(36),
unix_timestamp real,
dot_timestamp integer,
accel_x real,
accel_y real,
accel_z real
);
''')In color format, this looks like:
Above, we CREATE a table named free_acceleration with these columns:
id— a key to signify which row in the database it is. This isn’t 100% necessary, but reduces headaches for large datasets.bluetooth_address— the UUID address of the Movella DOTunix_timestamp— the actual time that the data was receiveddot_timestamp— the timestamp from the DOT, which is made up by the device and doesn’t correlate with any human timezoneaccel_x— the X of the accelerometeraccel_y— the Y of the accelerometeraccel_z— the Z of the accelerometer
Having a few timestamps, IDs, and addresses will help us keep track of all of the data when a lot of signals are coming in in realtime. The unix_timestamp will help especially, since the DOT’s timestamps appear arbitrary. (A timestamp of 1343451821 as seen below looks close to but is not actually related to any real timezone or time-keeping standard.)
Add the above create_database_table() function near the top of your code and add a line to call it near the top of the main connect() function:
async def connect(address):
number = address[:3]
connection = start_database(f"movella_dot{number}.db")
database = connection.cursor()
create_database_table(database) # Add this lineThe full script with database creation looks like so:
# Create a database to save Free Acceleration data from multiple Movella DOTs
import asyncio
import numpy as np
import sqlite3
from bleak import BleakClient
measurement_char_uuid = "15172001-4947-11e9-8646-d663bd873d93"
short_payload_char_uuid = "15172004-4947-11e9-8646-d663bd873d93"
# Replace this with a list of your DOT's UUID addresses
# (or "MAC addresses" for Windows and Linux users)
addresses = [
"509808FF-ECFE-895D-C1FE-BE5AC5DB6204",
"338312FA-C3D1-183F-325A-0726AFDBEB78"
]
class NotificationHandler:
'''This class allows us to add the DOT's UUID address to the data that gets printed'''
def __init__(self, device_address):
self.device_address = device_address
def callback(self, sender, data):
free_acceleration = encode_free_acceleration(data)[0]
free_acceleration = str(free_acceleration)[1:-1]
print(f"{self.device_address} -- {free_acceleration}")
def start_database(filename):
conn = sqlite3.connect(filename)
return conn
def create_database_table(cursor):
cursor.execute('''
CREATE TABLE IF NOT EXISTS free_acceleration (
id integer PRIMARY KEY,
bluetooth_address char(36),
unix_timestamp real,
dot_timestamp integer,
accel_x real,
accel_y real,
accel_z real
);
''')
def encode_free_acceleration(bytes_):
data_segments = np.dtype([
('timestamp', np.uint32),
('x', np.float32),
('y', np.float32),
('z', np.float32),
('zero_padding', np.uint32)
])
formatted_data = np.frombuffer(bytes_, dtype=data_segments)
return formatted_data
async def connect(address):
number = address[:2]
connection = start_database(f"movella_dot{number}.db")
database = connection.cursor()
create_database_table(database)
nh = NotificationHandler(address)
# Connect to the DOT and stream data
async with BleakClient(address) as client:
print(f"Client connection to `{client.address}: {client.is_connected}")
# Subscribe to data notifications
await client.start_notify(short_payload_char_uuid, nh.callback)
# Set and turn on measurement mode
binary_message = b"\x01\x01\x06"
await client.write_gatt_char(measurement_char_uuid, binary_message, response=True)
# Stream data for 10 seconds
await asyncio.sleep(10.0)
async def main():
await asyncio.gather(*(connect(addr) for addr in addresses))
if __name__ == "__main__":
asyncio.run(main())After you run the script above, you can preview your new empty databases, movella_dot*.db, in a free app like SQLite Browser. If you open any of them, you’ll see that they have one empty table with all of the columns that we just made.
The last steps are to format the incoming data and save it to the databases. We’ll do this by putting a formatting function and a SQL statement in the same callback() function that prints the data.
Step 4: Format the data
Right now, the accelerometer data is in a string. We need to extract it and format it, so that we can write it to our SQLite databases.
The data looks like this, which is close to what we want:
But we’ll make a new formatting function and rewrite the callback() to prep the data for SQL.
Right now, we’re combining the address with the timestamp, and XYZ accelerometer data (and a zero for padding that the DOT happens to give us) as:
address -- timestamp, X, Y, Z, 0Let’s change our callback like so to extract the variables (and import time):
import time
...
def callback(self, sender, data):
free_acceleration = encode_free_acceleration(data)[0]
free_acceleration = str(free_acceleration)[1:-1]
dot_timestamp, x, y, z, zero = free_acceleration.split(",")
unix_timestamp = time.time()
print(f"{self.device_address}, {unix_timestamp}, {dot_timestamp}, {x}, {y}, {z}")We’re extracting the data into separate variables. And we’re printing it back out for now as a sanity check before giving it to SQL.
The data looks appropriately formatted. Now, we can write it to our database.
Step 5: Write the data to the databases
We’ll write each piece of accelerometer data we get to each database with the SQL INSERT command:
database.execute('''
INSERT INTO free_acceleration
(bluetooth_address, unix_timestamp, dot_timestamp, accel_x, accel_y, accel_z)
VALUES
({... we'll put our newly extracted variables here ...})
);
''')
And we’ll put this into a new function, write_to_database(), below.
Note that there is no id column, like we had in our CREATE TABLE statement. That’s because id is a random value automatically generated by SQLite for each row of data.
Create a new function write_to_database() with the following:
def write_to_database(database, address, unix_time, dot_time, x, y, z):
database.execute(f'''
INSERT INTO free_acceleration
(bluetooth_address, unix_timestamp, dot_timestamp, accel_x, accel_y, accel_z)
VALUES
(\"{address}\", \"{unix_time}\", \"{dot_time}\", \"{x}\", \"{y}\", \"{z}\");
''')
This takes the arguments to write_to_database() and formats that them into the VALUES (...) section of the INSERT command.
Note how we use f-strings to insert the {variables} into the SQL.
Now, we can put write_to_database() into our original callback() function, so that every piece of incoming data gets written to the appropriate database:
def callback(self, sender, data):
free_acceleration = encode_free_acceleration(data)[0]
free_acceleration = str(free_acceleration)[1:-1]
dot_timestamp, x, y, z, zero = free_acceleration.split(",")
unix_timestamp = time.time()
# This is the new line
write_to_database(self.database, self.device_address, unix_timestamp, dot_timestamp, x, y, z)
print(f"{self.device_address}, {unix_timestamp}, {dot_timestamp}, {x}, {y}, {z}")One extra thing is necessary. In order to give the NotificationHandler class (and thus the callback() function) access to the database, we have to store the database a self.database variable in the class. That means we have to write a way to give NotificationHandler the database when both are created.
We have to change two things:
- The
__init__function ofNotificationHandler, so that it can accept and store the database object - Give the
databasetonh = NotificationHandler()when we createnh
Change the top of the NotificationHandler class to the following:
class NotificationHandler:
'''This class allows us to add the DOT's UUID address to the data that gets printed'''
def __init__(self, device_address, database):
self.device_address = device_address
self.database = database # Now we can use the database in callback()And change the nh = NotificationHandler(...) part to:
async def connect(address):
number = address[:2]
connection = start_database(f"movella_dot{number}.db")
database = connection.cursor()
create_database_table(database)
# Add `database` as a second argument here
nh = NotificationHandler(address, database)This changes the full script to:
# Stream Free Acceleration data from multiple Movella DOTs
import asyncio
import numpy as np
import sqlite3
import time
from bleak import BleakClient
measurement_char_uuid = "15172001-4947-11e9-8646-d663bd873d93"
short_payload_char_uuid = "15172004-4947-11e9-8646-d663bd873d93"
# Replace this with a list of your DOT's UUID addresses
# (or "MAC addresses" for Windows and Linux users)
addresses = [
"509808FF-ECFE-895D-C1FE-BE5AC5DB6204",
"338312FA-C3D1-183F-325A-0726AFDBEB78"
]
class NotificationHandler:
'''This class allows us to add the DOT's UUID address to the data that gets printed'''
def __init__(self, device_address, database):
self.device_address = device_address
self.database = database
def callback(self, sender, data):
free_acceleration = encode_free_acceleration(data)[0]
free_acceleration = str(free_acceleration)[1:-1]
dot_timestamp, x, y, z, zero = free_acceleration.split(",")
unix_timestamp = time.time()
write_to_database(self.database, self.device_address, unix_timestamp, dot_timestamp, x, y, z)
print(f"{self.device_address}, {unix_timestamp}, {dot_timestamp}, {x}, {y}, {z}")
def start_database(filename):
conn = sqlite3.connect(filename)
return conn
def create_database_table(cursor):
cursor.execute('''
CREATE TABLE IF NOT EXISTS free_acceleration (
id integer PRIMARY KEY,
bluetooth_address char(36),
unix_timestamp real,
dot_timestamp integer,
accel_x real,
accel_y real,
accel_z real
);
''')
def write_to_database(database, address, unix_time, dot_time, x, y, z):
database.execute(f'''
INSERT INTO free_acceleration
(bluetooth_address, unix_timestamp, dot_timestamp, accel_x, accel_y, accel_z)
VALUES
(\"{address}\", \"{unix_time}\", \"{dot_time}\", \"{x}\", \"{y}\", \"{z}\");
''')
def encode_free_acceleration(bytes_):
data_segments = np.dtype([
('timestamp', np.uint32),
('x', np.float32),
('y', np.float32),
('z', np.float32),
('zero_padding', np.uint32)
])
formatted_data = np.frombuffer(bytes_, dtype=data_segments)
return formatted_data
async def connect(address):
number = address[:2]
connection = start_database(f"movella_dot{number}.db")
database = connection.cursor()
create_database_table(database)
nh = NotificationHandler(address, database)
# Connect to the DOT and stream data
async with BleakClient(address) as client:
print(f"Client connection to `{client.address}: {client.is_connected}")
# Subscribe to data notifications
await client.start_notify(short_payload_char_uuid, nh.callback)
# Set and turn on measurement mode
binary_message = b"\x01\x01\x06"
await client.write_gatt_char(measurement_char_uuid, binary_message, response=True)
# Stream data for 10 seconds
await asyncio.sleep(10.0)
async def main():
await asyncio.gather(*(connect(addr) for addr in addresses))
if __name__ == "__main__":
asyncio.run(main())This will start sending data to our databases! However, we need one last step: Saving the data with SQLite’s commit() function.
Step 6: Save the data to the databases
If we run the script from Step 5 and open one of our movella_dot*.db files in SQLite Browser, we won’t see any actual data.
That’s because we didn’t save the database with commit().
To fix this, at the very end of async def connect() , add a single connection.commit():
async def connect(address):
...
async with BleakClient(address) as client:
...
connection.commit() # Add this lineThis will save all the data in our databases. Make sure it is in the connect() function but outside of the BleakClient() as client block.
If we run the script again and open one our movella_dot*.db files, we’ll see our saved DOT data.
We have the DOT’s UUID address, the unix timestamp, the DOT’s timestamp, and the accelerometer’s X, Y, and Z values.
That is all that is needed to get your Movella/Xsens DOT data into SQLite. There were actually a lot of weird quirks to overcome with code. Thanks for sticking it out and reading.
Questions and Feedback
If you have questions or feedback, email us at [email protected] or message us on Instagram (@protobioengineering).
If you found this article useful, you can help us complete more content like this by donating a coffee!
The Full Script
# Stream Free Acceleration data from multiple Movella DOTs
import asyncio
import numpy as np
import sqlite3
import time
from bleak import BleakClient
measurement_char_uuid = "15172001-4947-11e9-8646-d663bd873d93"
short_payload_char_uuid = "15172004-4947-11e9-8646-d663bd873d93"
# Replace this with a list of your DOT's UUID addresses
# (or "MAC addresses" for Windows and Linux users)
addresses = [
"509808FF-ECFE-895D-C1FE-BE5AC5DB6204",
"338312FA-C3D1-183F-325A-0726AFDBEB78"
]
class NotificationHandler:
'''This class allows us to add the DOT's UUID address to the data that gets printed'''
def __init__(self, device_address, database):
self.device_address = device_address
self.database = database
def callback(self, sender, data):
free_acceleration = encode_free_acceleration(data)[0]
free_acceleration = str(free_acceleration)[1:-1]
dot_timestamp, x, y, z, zero = free_acceleration.split(",")
unix_timestamp = time.time()
write_to_database(self.database, self.device_address, unix_timestamp, dot_timestamp, x, y, z)
print(f"{self.device_address}, {unix_timestamp}, {dot_timestamp}, {x}, {y}, {z}")
def start_database(filename):
conn = sqlite3.connect(filename)
return conn
def create_database_table(cursor):
cursor.execute('''
CREATE TABLE IF NOT EXISTS free_acceleration (
id integer PRIMARY KEY,
bluetooth_address char(36),
unix_timestamp real,
dot_timestamp integer,
accel_x real,
accel_y real,
accel_z real
);
''')
def write_to_database(database, address, unix_time, dot_time, x, y, z):
database.execute(f'''
INSERT INTO free_acceleration
(bluetooth_address, unix_timestamp, dot_timestamp, accel_x, accel_y, accel_z)
VALUES
(\"{address}\", \"{unix_time}\", \"{dot_time}\", \"{x}\", \"{y}\", \"{z}\");
''')
def encode_free_acceleration(bytes_):
data_segments = np.dtype([
('timestamp', np.uint32),
('x', np.float32),
('y', np.float32),
('z', np.float32),
('zero_padding', np.uint32)
])
formatted_data = np.frombuffer(bytes_, dtype=data_segments)
return formatted_data
async def connect(address):
number = address[:2]
connection = start_database(f"movella_dot{number}.db")
database = connection.cursor()
create_database_table(database)
nh = NotificationHandler(address, database)
# Connect to the DOT and stream data
async with BleakClient(address) as client:
print(f"Client connection to `{client.address}: {client.is_connected}")
# Subscribe to data notifications
await client.start_notify(short_payload_char_uuid, nh.callback)
# Set and turn on measurement mode
binary_message = b"\x01\x01\x06"
await client.write_gatt_char(measurement_char_uuid, binary_message, response=True)
# Stream data for 10 seconds
await asyncio.sleep(10.0)
connection.commit()
async def main():
await asyncio.gather(*(connect(addr) for addr in addresses))
if __name__ == "__main__":
asyncio.run(main())Related Resources
- Movella Support and Knowledge Base
- Movella SDKs and Bluetooth Documentation
- AsyncIO in Python: A Complete Walkthrough
Questions and Feedback
If you have questions or feedback, email us at [email protected] or message us on Instagram (@protobioengineering).
If you liked this article, consider supporting us by donating a coffee.