# vibration sensor and temperature sensor code backup:
```python=
from adxl345 import ADXL345
import time
import os
import paho.mqtt.client as mqtt
import threading
from scipy import pi
import matplotlib.pyplot as plt
os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')
sensoid= ['3b-0000001817de','3b-0000001817d1']
global temp_sensor
global temperature
temperature=[0,0]
def on_connect(client, userdata, flags, rc):
print("Connected with result code "+str(rc))
client.subscribe("IOT_NEAT_TOPIC01")
def on_message(client, userdata, msg):
print(msg.topic+" "+str(msg.payload))
def temp_raw(temp_sensor):
#self.temp_sensor=temp_sensor
f = open(temp_sensor,'r')
lines = f.readlines()
f.close()
return lines
def read_temp(temp_sensor):
lines = temp_raw(temp_sensor)
while lines[0].strip()[-3: ] != 'YES':
time.sleep(0.01)
lines = temp_raw()
temp_output = lines[1].find('t=')
if temp_output != -1:
temp_string = lines[1].strip()[temp_output+2:]
temp_c = float(temp_string)/1000.0
#temp_f = temp_c * 9.0/5.0+32.0
return temp_c
def thread_sensor1():
while 1:
global x_1
global y_1
global z_1
adxl345 = ADXL345()
adxl345.address = 0x53
x_1,y_1,z_1 = adxl345.read()
axes = adxl345.getAxes(True)
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print('X={0}, Y={1}, Z={2}'.format(x_1, y_1, z_1))
time.sleep(0.1)
def thread_sensor2():
while 1:
global x_2
global y_2
global z_2
adxl345_1 = ADXL345()
adxl345_1.address = 0x04
x_2,y_2,z_2=adxl345_1.read()
axes_1 = adxl345_1.getAxes(True)
print ("ADXL345_1 on address 0x%x:" % (adxl345_1.address))
print('X={0}, Y={1}, Z={2}'.format(x_2, y_2, z_2))
mqttd.publish(_g_cst_MQTTTopicName, 'X_1={0}, Y_1={1}, Z_1={2}'.format(x_2, y_2, z_2))
time.sleep(0.1)
def thread_mqtt():
mqttc = mqtt.Client("python_pub")
_g_cst_ToMQTTTopicServerIP = "35.229.37.146"
_g_cst_ToMQTTTopicServerPort = 1883 #port
_g_cst_MQTTTopicName = "IOT_NEAT_TOPIC01" #TOPIC name
mqttc.connect(_g_cst_ToMQTTTopicServerIP, _g_cst_ToMQTTTopicServerPort)
while 1:
#x,y,z = accel.read()
#mqttc.publish(_g_cst_MQTTTopicName, 'X_1={0}, Y_1={1}, Z_1={2},X_2 = {3},Y_2 = {4},Z_2={5}'.format(x_1, y_1, z_1, x_2,y_2,z_2))
mqttc.publish(_g_cst_MQTTTopicName, 'X_1={0}, Y_1={1}, Z_1={2},temp_1 = {3},temp_2={4}'.format(x_1,y_1,z_1,temperature[0],temperature[1]))
time.sleep(0.1)
def thread_temperature():
while True:
for sensor in range(len(sensoid)):
temp_sensor = '/sys/bus/w1/devices/'+sensoid[sensor]+'/w1_slave'
temperature[sensor] = read_temp(temp_sensor)
print('the %d temp sensor is '%(sensor+1),temperature[sensor])
time.sleep(0.01)
"""
while 1:
adxl345 = ADXL345()
adxl345.address = 0x53
x_1,y_1,z_1 = adxl345.read()
axes = adxl345.getAxes(True)
adxl345_1 = ADXL345()
adxl345_1.address = 0x04
axes_1 = adxl345_1.getAxes(True)
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print (" x = %.3fG" % ( axes['x'] ))
print (" y = %.3fG" % ( axes['y'] ))
print (" z = %.3fG" % ( axes['z'] ))
print('X={0}, Y={1}, Z={2}'.format(x_1, y_1, z_1))
time.sleep(0.1)
print ("ADXL345_1 on address 0x%x:" % (adxl345_1.address))
print (" x = %.3fG" % ( axes_1['x'] ))
print (" y = %.3fG" % ( axes_1['y'] ))
print (" z = %.3fG" % ( axes_1['z'] ))
print('X={0}, Y={1}, Z={2}'.format(x_2, y_2, z_2))
time.sleep(0.1)
"""
client = mqtt.Client()
#client_1 = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
#client_1.on_connect = on_connect
#client_1.on_message = on_message
client.connect("35.229.37.146", 1883, 60)
#client_1.connect("35.229.37.146",1884,58)
my_thread1 = threading.Thread(target = thread_sensor1)
#my_thread2 = threading.Thread(target = thread_sensor2)
my_thread3 = threading.Thread(target = thread_mqtt)
#my_thread3 = threading.Thread(target = thread_clock)
my_thread4 = threading.Thread(target = thread_temperature)
my_thread1.start()
#my_thread2.start()
my_thread3.start()
my_thread4.start()
print("this is main thread") # should not go here
client.loop_forever()
```
# temperature:
```python=
import os
import time
os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')
#temp_sensor = '/sys/bus/w1/devices/'
sensoid= ['3b-0000001817de','3b-0000001817d1']
temperature=[0,0]
def temp_raw():
f = open(temp_sensor,'r')
lines = f.readlines()
f.close()
return lines
def read_temp():
lines = temp_raw()
while lines[0].strip()[-3: ] != 'YES':
time.sleep(0.05)
lines = temp_raw()
temp_output = lines[1].find('t=')
if temp_output != -1:
temp_string = lines[1].strip()[temp_output+2:]
temp_c = float(temp_string)/1000.0
#temp_f = temp_c * 9.0/5.0+32.0
return temp_c
while True:
for sensor in range(len(sensoid)):
temp_sensor = '/sys/bus/w1/devices/'+sensoid[sensor]+'/w1_slave'
temperature[sensor] = read_temp()
print('the %d temp sensor is '%(sensor+1),temperature[sensor])
time.sleep(0.05)
```
[low-pass filter](https://howtomechatronics.com/tutorials/arduino/how-to-track-orientation-with-arduino-and-adxl345-accelerometer/)
[Moving Average Filters](https://www.analog.com/media/en/technical-documentation/dsp-book/dsp_book_Ch15.pdf)
```python=
from adxl345 import ADXL345
import time
import os
import threading
from scipy import pi
import matplotlib.pyplot as plt
import numpy as np
import copy
os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')
sensoid= ['3b-0000001817de','3b-0000001817d1']
global temp_sensor
global temperature
global x_1
global y_1
global z_1
global x_tmp
x_tmp = 0
temperature=[0,0]
def temp_raw(temp_sensor):
#self.temp_sensor=temp_sensor
f = open(temp_sensor,'r')
lines = f.readlines()
f.close()
return lines
def read_temp(temp_sensor):
lines = temp_raw(temp_sensor)
while lines[0].strip()[-3: ] != 'YES':
time.sleep(0.01)
lines = temp_raw()
temp_output = lines[1].find('t=')
if temp_output != -1:
temp_string = lines[1].strip()[temp_output+2:]
temp_c = float(temp_string)/1000.0
#temp_f = temp_c * 9.0/5.0+32.0
return temp_c
def thread_sensor1():
i=0
k=0
global x_2
global y_2
global z_2
global velocity_x
x_2 = 0
y_2 = 0
z_2 = 0
valocity_x = 0
y_tmp = 0
x_tmp = 0
z_tmp = 0
x_3 = 0
vib_data_x = np.zeros(5)
vib_data_y = np.zeros(5)
vib_data_z = np.zeros(5)
while 1:
adxl345 = ADXL345()
adxl345.address = 0x53
adxl345.setoffset_x(0xFF)
adxl345.setoffset_y(0x02)
adxl345.setoffset_z(0x06)
adxl345.setRange(0x03)
#x_1,y_1,z_1 = adxl345.read()
axes = adxl345.getAxes_in_mm(gforce = True)
x_1 = abs(axes['x'])
y_1 = abs(axes['y'])
z_1 = abs(axes['z'])
vib_data_y[i]= y_1-4
vib_data_x[i] = x_1-12
velocity_x = velocity_x + vib_data_x[i]*0.1
vib_data_z[i] = z_1
#print('vib_data_x_i is %d'%(vib_data_x[i]))
#print(i)
i+=1
if i==5:
i=0
for j in range(0,5):
x_tmp = vib_data_x[j]+x_tmp
y_tmp = vib_data_y[j]+y_tmp
z_tmp = vib_data_z[j]+z_tmp
#print('vib_data_x is %d'%(vib_data_x[j]))
vib_data_x[j]=0
vib_data_y[j]=0
vib_data_z[j]=0
x_2 = abs((x_tmp.copy())/5)
y_2 = abs((y_tmp.copy())/5)
z_2 = abs((z_tmp.copy())/5)
x_tmp=0
y_tmp=0
z_tmp=0
#print ("ADXL345 on address 0x%x:" % (adxl345.address))
print('the temperature are : ')
print('temp_1 = {0},temp_2={1}'.format(temperature[0],temperature[1]))
print(' ')
print('the vibration is :')
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print (" x = %.4fmm/s^2" % (x_2))
print (" y = %.4fmm/s^2" % (y_2))
print (" z = %.4fmm/s^2" % (z_2))
print(' ')
time.sleep(0.1)
def thread_sensor2():
while 1:
global x_2
global y_2
global z_2
adxl345_1 = ADXL345()
adxl345_1.address = 0x04
x_2,y_2,z_2=adxl345_1.read()
axes_1 = adxl345_1.getAxes(True)
print ("ADXL345_1 on address 0x%x:" % (adxl345_1.address))
print('X={0}, Y={1}, Z={2}'.format(x_2, y_2, z_2))
time.sleep(0.1)
def thread_temperature():
while True:
for sensor in range(len(sensoid)):
temp_sensor = '/sys/bus/w1/devices/'+sensoid[sensor]+'/w1_slave'
temperature[sensor] = read_temp(temp_sensor)
#print('the %d temp sensor is '%(sensor+1),temperature[sensor])
"""
while 1:
adxl345 = ADXL345()
adxl345.address = 0x53
x_1,y_1,z_1 = adxl345.read()
axes = adxl345.getAxes(True)
adxl345_1 = ADXL345()
adxl345_1.address = 0x04
axes_1 = adxl345_1.getAxes(True)
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print (" x = %.3fG" % ( axes['x'] ))
print (" y = %.3fG" % ( axes['y'] ))
print (" z = %.3fG" % ( axes['z'] ))
print('X={0}, Y={1}, Z={2}'.format(x_1, y_1, z_1))
time.sleep(0.1)
print ("ADXL345_1 on address 0x%x:" % (adxl345_1.address))
print (" x = %.3fG" % ( axes_1['x'] ))
print (" y = %.3fG" % ( axes_1['y'] ))
print (" z = %.3fG" % ( axes_1['z'] ))
print('X={0}, Y={1}, Z={2}'.format(x_2, y_2, z_2))
time.sleep(0.1)
"""
my_thread1 = threading.Thread(target = thread_sensor1)
#my_thread2 = threading.Thread(target = thread_sensor2)
#my_thread3 = threading.Thread(target = thread_mqtt)
#my_thread3 = threading.Thread(target = thread_print)
my_thread4 = threading.Thread(target = thread_temperature)
my_thread1.start()
#my_thread2.start()
#my_thread3.start()
my_thread4.start()
print("this is main thread") # should not go here
```
velocity = velocity + vib_data_x* time (0.5s)
# ADXL345:
```python=
# ADXL345 Python library for Raspberry Pi
#
# author: Jonathan Williamson
# license: BSD, see LICENSE.txt included in this package
#
# This is a Raspberry Pi Python implementation to help you get started with
# the Adafruit Triple Axis ADXL345 breakout board:
# http://shop.pimoroni.com/products/adafruit-triple-axis-accelerometer
import struct
import smbus
from time import sleep
import Adafruit_GPIO.I2C as I2C
# select the correct i2c bus for this revision of Raspberry Pi
revision = ([l[12:-1] for l in open('/proc/cpuinfo','r').readlines() if l[:8]=="Revision"]+['0000'])[0]
bus = smbus.SMBus(1)
# ADXL345 constants
EARTH_GRAVITY_MS2 = 9.80665
SCALE_MULTIPLIER = 0.004
ADXL345_REG_DATAX0 = 0x32 # X-axis data 0 (6 bytes for X/Y/Z)
DATA_FORMAT = 0x31
BW_RATE = 0x2C
POWER_CTL = 0x2D
BW_RATE_1600HZ = 0x0F
BW_RATE_800HZ = 0x0E
BW_RATE_400HZ = 0x0D
BW_RATE_200HZ = 0x0C
BW_RATE_100HZ = 0x0B
BW_RATE_50HZ = 0x0A
BW_RATE_25HZ = 0x09
RANGE_2G = 0x00
RANGE_4G = 0x01
RANGE_8G = 0x02
RANGE_16G = 0x03
MEASURE = 0x08
AXES_DATA = 0x32
ADXL345_REG_DEVID = 0x00 # Device ID
ADXL345_REG_POWER_CTL = 0x2D
Reg_OFSX = 0x1E
Reg_OFSY = 0x1F
Reg_OFSZ = 0x20
class ADXL345:
address = None
def __init__(self,i2c=None,address=0x53,**kwargs):
if i2c is None:
import Adafruit_GPIO.I2C as I2C
i2c = I2C
self.address = address
self.setBandwidthRate(BW_RATE_50HZ)
self.setRange(RANGE_2G)
self.enableMeasurement()
self._device = i2c.get_i2c_device(address, **kwargs)
if self._device.readU8(ADXL345_REG_DEVID) == 0xE5:
self._device.write8(ADXL345_REG_POWER_CTL, 0x08)
def enableMeasurement(self):
bus.write_byte_data(self.address, POWER_CTL, MEASURE)
def setBandwidthRate(self, rate_flag):
bus.write_byte_data(self.address, BW_RATE, rate_flag)
# set the measurement range for 10-bit readings
def setRange(self, range_flag):
value = bus.read_byte_data(self.address, DATA_FORMAT)
value &= ~0x0F;
value |= range_flag;
value |= 0x08;
bus.write_byte_data(self.address, DATA_FORMAT, value)
# returns the current reading from the sensor for each axis
#
# parameter gforce:
# False (default): result is returned in m/s^2
# True : result is returned in gs
def getAxes(self, gforce = False):
bytes = bus.read_i2c_block_data(self.address, AXES_DATA, 6)
x = bytes[0] | (bytes[1] << 8)
if(x & (1 << 16 - 1)):
x = x - (1<<16)
y = bytes[2] | (bytes[3] << 8)
if(y & (1 << 16 - 1)):
y = y - (1<<16)
z = bytes[4] | (bytes[5] << 8)
if(z & (1 << 16 - 1)):
z = z - (1<<16)
x = x * SCALE_MULTIPLIER
y = y * SCALE_MULTIPLIER
z = z * SCALE_MULTIPLIER
if gforce == False:
x = x * EARTH_GRAVITY_MS2
y = y * EARTH_GRAVITY_MS2
z = z * EARTH_GRAVITY_MS2
x = round(x, 4)
y = round(y, 4)
z = round(z, 4)
return {"x": x, "y": y, "z": z}
def getAxes_in_mm(self, gforce = False):
bytes = bus.read_i2c_block_data(self.address, AXES_DATA, 6)
x_1 = bytes[0] | (bytes[1] << 8)
if(x_1 & (1 << 16 - 1)):
x_1 = x_1 - (1<<16)
y_1 = bytes[2] | (bytes[3] << 8)
if(y_1 & (1 << 16 - 1)):
y_1 = y_1 - (1<<16)
z_1 = bytes[4] | (bytes[5] << 8)
if(z_1 & (1 << 16 - 1)):
z_1 = z_1 - (1<<16)
x_1= x_1 * SCALE_MULTIPLIER*1000
y_1 = y_1 * SCALE_MULTIPLIER*1000
z_1= z_1 * SCALE_MULTIPLIER*1000
if gforce == False:
x_1 = x_1 * EARTH_GRAVITY_MS2*1000
y_1 = y_1 * EARTH_GRAVITY_MS2*1000
z_1 = z_1 * EARTH_GRAVITY_MS2*1000
x_1 = round(x_1, 4)
y_1 = round(y_1, 4)
z_1 = round(z_1, 4)
return {"x": x_1, "y": y_1, "z": z_1}
def read(self):
raw = self._device.readList(ADXL345_REG_DATAX0, 6)
return struct.unpack('<hhh', raw)
def setoffset_x(self,value):
bus.write_byte_data(self.address, Reg_OFSX, value)
def setoffset_y(self,value2):
bus.write_byte_data(self.address, Reg_OFSY, value2)
def setoffset_z(self,value3):
bus.write_byte_data(self.address, Reg_OFSZ, value3)
if __name__ == "__main__":
# if run directly we'll just create an instance of the class and output
# the current readings
adxl345 = ADXL345()
#adxl345.setoffset_x(0xFF)
#adxl345.setoffset_y(0x02)
#adxl345.setoffset_z(0x09)
axes= adxl345.getAxes(True)
x, y, z= adxl345.read()
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print (" x = %.3fG" % ( axes['x'] ))
print (" y = %.3fG" % ( axes['y'] ))
print (" z = %.3fG" % ( axes['z'] ))
print('X={0}, Y={1}, Z={2}'.format(x, y, z))
```
# test_offline(acceleration)
```python=
from adxl345 import ADXL345
import time
import os
import threading
from scipy import pi
import matplotlib.pyplot as plt
import numpy as np
import copy
os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')
sensoid= ['3b-0000001817de','3b-0000001817d1']
global temp_sensor
global temperature
global x_1
global y_1
global z_1
global x_tmp
x_tmp = 0
temperature=[0,0]
def temp_raw(temp_sensor):
#self.temp_sensor=temp_sensor
f = open(temp_sensor,'r')
lines = f.readlines()
f.close()
return lines
def read_temp(temp_sensor):
lines = temp_raw(temp_sensor)
while lines[0].strip()[-3: ] != 'YES':
time.sleep(0.01)
lines = temp_raw()
temp_output = lines[1].find('t=')
if temp_output != -1:
temp_string = lines[1].strip()[temp_output+2:]
temp_c = float(temp_string)/1000.0
#temp_f = temp_c * 9.0/5.0+32.0
return temp_c
def thread_sensor1():
i=0
k=0
global x_2
global y_2
global z_2
x_2 = 0
y_2 = 0
z_2 = 0
y_tmp = 0
x_tmp = 0
z_tmp = 0
vib_data_x = np.zeros(40)
vib_data_y = np.zeros(40)
vib_data_z = np.zeros(40)
adxl345 = ADXL345()
adxl345.address = 0x53
adxl345.setoffset_x(0x00) # 0x00 0xFF 0X00 0XFF 0X00 0xFF 0XFF
adxl345.setoffset_y(0x00) # 0x02 0x02 0X02 0X03 0X03 0X03 0X01
adxl345.setoffset_z(0x06) # 0XFE 0XF8 0XFC 0XFC 0XFE 0X03 0X03
adxl345.setRange(0x03) # 0X01 0X01 0X03 0X03 0X03 0X01 0X01
while 1:
#x_1,y_1,z_1 = adxl345.read()
axes = adxl345.getAxes_in_mm(gforce = True)
x_1 = (axes['x'])
y_1 = (axes['y'])
z_1 = abs(axes['z'])
vib_data_y[i]= y_1
vib_data_x[i] = x_1
vib_data_z[i] = z_1
#print('vib_data_x_i is %d'%(vib_data_x[i]))
#print(x_1)
i+=1
if i==40:
i=0
for j in range(0,40):
x_tmp = vib_data_x[j]+x_tmp
#print('test is %d '%(j),vib_data_x[j])
y_tmp = vib_data_y[j]+y_tmp
z_tmp = vib_data_z[j]+z_tmp
#print('vib_data_x is %d'%(vib_data_x[j]))
vib_data_x[j]=0
vib_data_y[j]=0
vib_data_z[j]=0
x_2 = ((x_tmp.copy())/40)
y_2 = ((y_tmp.copy())/40)
z_2 = abs((z_tmp.copy())/40)
x_tmp=0
y_tmp=0
z_tmp=0
#print ("ADXL345 on address 0x%x:" % (adxl345.address))
print('the temperature are : ')
print('temp_1 = {0},temp_2={1}'.format(temperature[0],temperature[1]))
print(' ')
print('the vibration is :')
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print (" x = %.4fmm/s^2" % (x_2))
print (" y = %.4fmm/s^2" % (y_2))
print (" z = %.4fmm/s^2" % (z_2))
print(' ')
time.sleep(0.05)
def thread_sensor2():
while 1:
global x_2
global y_2
global z_2
adxl345_1 = ADXL345()
adxl345_1.address = 0x04
x_2,y_2,z_2=adxl345_1.read()
axes_1 = adxl345_1.getAxes(True)
print ("ADXL345_1 on address 0x%x:" % (adxl345_1.address))
print('X={0}, Y={1}, Z={2}'.format(x_2, y_2, z_2))
time.sleep(0.1)
def thread_temperature():
while True:
for sensor in range(len(sensoid)):
temp_sensor = '/sys/bus/w1/devices/'+sensoid[sensor]+'/w1_slave'
temperature[sensor] = read_temp(temp_sensor)
#print('the %d temp sensor is '%(sensor+1),temperature[sensor])
"""
while 1:
adxl345 = ADXL345()
adxl345.address = 0x53
x_1,y_1,z_1 = adxl345.read()
axes = adxl345.getAxes(True)
adxl345_1 = ADXL345()
adxl345_1.address = 0x04
axes_1 = adxl345_1.getAxes(True)
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print (" x = %.3fG" % ( axes['x'] ))
print (" y = %.3fG" % ( axes['y'] ))
print (" z = %.3fG" % ( axes['z'] ))
print('X={0}, Y={1}, Z={2}'.format(x_1, y_1, z_1))
time.sleep(0.1)
print ("ADXL345_1 on address 0x%x:" % (adxl345_1.address))
print (" x = %.3fG" % ( axes_1['x'] ))
print (" y = %.3fG" % ( axes_1['y'] ))
print (" z = %.3fG" % ( axes_1['z'] ))
print('X={0}, Y={1}, Z={2}'.format(x_2, y_2, z_2))
time.sleep(0.1)
"""
my_thread1 = threading.Thread(target = thread_sensor1)
#my_thread2 = threading.Thread(target = thread_sensor2)
#my_thread3 = threading.Thread(target = thread_mqtt)
#my_thread3 = threading.Thread(target = thread_print)
my_thread4 = threading.Thread(target = thread_temperature)
my_thread1.start()
#my_thread2.start()
#my_thread3.start()
my_thread4.start()
print("this is main thread") # should not go here
```
# test_off line(velocity)
```python=
from adxl345 import ADXL345
import time
import os
import threading
from scipy import pi
import matplotlib.pyplot as plt
import numpy as np
import copy
os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')
sensoid= ['3b-0000001817de','3b-0000001817d1']
global temp_sensor
global temperature
global x_1
global y_1
global z_1
global x_tmp
x_tmp = 0
temperature=[0,0]
def temp_raw(temp_sensor):
#self.temp_sensor=temp_sensor
f = open(temp_sensor,'r')
lines = f.readlines()
f.close()
return lines
def read_temp(temp_sensor):
lines = temp_raw(temp_sensor)
while lines[0].strip()[-3: ] != 'YES':
time.sleep(0.01)
lines = temp_raw()
temp_output = lines[1].find('t=')
if temp_output != -1:
temp_string = lines[1].strip()[temp_output+2:]
temp_c = float(temp_string)/1000.0
#temp_f = temp_c * 9.0/5.0+32.0
return temp_c
def thread_sensor1():
i=0
k=0
global x_2
global velocity_x
global velocity_y
global velocity_z
global y_2
global z_2
x_2 = 0
y_2 = 0
z_2 = 0
velocity_x = 0
velocity_y = 0
velocity_z = 0
#y_tmp = 0
velocity_x_tmp = 0
velocity_x_tmp_1 = [0,0]
velocity_y_tmp = 0
velocity_z_tmp = 0
#x_tmp = 0
z_tmp = 0
vib_data_x = np.zeros(40)
vib_data_y = np.zeros(40)
vib_data_z = np.zeros(40)
adxl345 = ADXL345()
adxl345.address = 0x53
adxl345.setoffset_x(0xFC)
adxl345.setoffset_y(0x01)
adxl345.setoffset_z(0x06)
adxl345.setRange(0x03)
while 1:
#x_1,y_1,z_1 = adxl345.read()
axes = adxl345.getAxes_in_mm(gforce = True)
x_1 = abs(axes['x'])
y_1 = abs(axes['y'])
z_1 = (axes['z'])
vib_data_y[i]= y_1
vib_data_x[i] = x_1 # x acceleration
vib_data_z[i] = z_1
#print('vib_data_x_i is %d'%(vib_data_x[i]))
#print(z_1)
i+=1
if i==40:
i=0
for j in range(0,40):
velocity_x_tmp = vib_data_x[j]*0.02+velocity_x_tmp
velocity_y_tmp = vib_data_y[j]*0.02+velocity_y_tmp
velocity_z_tmp = vib_data_z[j]*0.02+velocity_z_tmp
#y_tmp = vib_data_y[j]+y_tmp
z_tmp = vib_data_z[j]+z_tmp
#print('vib_data_x is %d'%(vib_data_x[j]))
vib_data_x[j]=0
vib_data_y[j]=0
vib_data_z[j]=0
z_2 = abs((z_tmp.copy())/40)
velocity_x = ((velocity_x_tmp.copy())/40)
velocity_y = ((velocity_y_tmp.copy())/40)
velocity_z = abs((velocity_z_tmp.copy())/40)
z_tmp=0
velocity_x_tmp = 0
velocity_y_tmp = 0
#print ("ADXL345 on address 0x%x:" % (adxl345.address))
print('******')
print('the temperature are : ')
print('temp_1 = {0},temp_2={1}'.format(temperature[0],temperature[1]))
print('******')
print('the vibration is :')
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print("velocity_x = %.4fmm/s" % (velocity_x))
print("velocity_y = %.4fmm/s" % (velocity_y))
print ("z = %.4fmm/s^2" % (z_2))
#print(" velocity_z = %.4fmm/s" % (velocity_z))
time.sleep(0.02)
def thread_sensor2():
while 1:
global x_2
global y_2
global z_2
adxl345_1 = ADXL345()
adxl345_1.address = 0x04
x_2,y_2,z_2=adxl345_1.read()
axes_1 = adxl345_1.getAxes(True)
print ("ADXL345_1 on address 0x%x:" % (adxl345_1.address))
print('X={0}, Y={1}, Z={2}'.format(x_2, y_2, z_2))
time.sleep(0.1)
def thread_temperature():
while True:
for sensor in range(len(sensoid)):
temp_sensor = '/sys/bus/w1/devices/'+sensoid[sensor]+'/w1_slave'
temperature[sensor] = read_temp(temp_sensor)
#print('the %d temp sensor is '%(sensor+1),temperature[sensor])
my_thread1 = threading.Thread(target = thread_sensor1)
#my_thread2 = threading.Thread(target = thread_sensor2)
#my_thread3 = threading.Thread(target = thread_mqtt)
#my_thread3 = threading.Thread(target = thread_print)
my_thread4 = threading.Thread(target = thread_temperature)
my_thread1.start()
#my_thread2.start()
#my_thread3.start()
my_thread4.start()
print("this is main thread") # should not go here
```
# Multiple vibration sensors:
## 校正
```python=
from adxl345 import ADXL345
from Adafruit_GPIO import I2C
import time
import os
import threading
from scipy import pi
import matplotlib.pyplot as plt
import numpy as np
import copy
os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')
sensoid= ['3b-0000001817de','3b-0000001817d1']
global temp_sensor
global temperature
global x_1
global y_1
global z_1
global x_tmp
global size
size = 20
x_tmp = 0
temperature=[0,0]
def temp_raw(temp_sensor):
#self.temp_sensor=temp_sensor
f = open(temp_sensor,'r')
lines = f.readlines()
f.close()
return lines
def read_temp(temp_sensor):
lines = temp_raw(temp_sensor)
while lines[0].strip()[-3: ] != 'YES':
time.sleep(0.01)
lines = temp_raw()
temp_output = lines[1].find('t=')
if temp_output != -1:
temp_string = lines[1].strip()[temp_output+2:]
temp_c = float(temp_string)/1000.0
#temp_f = temp_c * 9.0/5.0+32.0
return temp_c
tca = I2C.get_i2c_device(address=0x70)
def tca_select(channel):
if channel > 7:
return
tca.writeRaw8(1 << channel)
def tca_set(mask):
if mask > 0xff:
return
tca.writeRaw8(mask)
def thread_sensor1():
i=0
k=0
global x_2
global velocity_x
global velocity_y
global velocity_z
global y_2
global z_2
x_2 = 0
y_2 = 0
z_2 = 0
velocity_x = 0
velocity_y = 0
velocity_z = 0
#y_tmp = 0
velocity_x_tmp = 0
velocity_y_tmp = 0
velocity_z_tmp = 0
#x_tmp = 0
z_tmp = 0
vib_data_x = np.zeros(size)
vib_data_y = np.zeros(size)
vib_data_z = np.zeros(size)
adxl345 = ADXL345()
adxl345.address = 0x53# 0XFC
adxl345.setoffset_z(0x06)
adxl345.setRange(0x03)
print('test')
while 1:
tca_select(0)
adxl345.setoffset_x(0x01)
adxl345.setoffset_y(0x01) # 0X01
#x_1,y_1,z_1 = adxl345.read()
axes = adxl345.getAxes_in_mm(gforce = True)
z_1 = (axes['z'])
vib_data_y[i]= (axes['y'])
vib_data_x[i] = (axes['x']) # x acceleration
vib_data_z[i] = z_1
#print('vib_data_x_i is %d'%(vib_data_x[i]))
#print(z_1)
i+=1
if i==size:
i=0
for j in range(0,size):
velocity_x_tmp = vib_data_x[j]+velocity_x_tmp
velocity_y_tmp = vib_data_y[j]+velocity_y_tmp
#y_tmp = vib_data_y[j]+y_tmp
#z_tmp = vib_data_z[j]+z_tmp
#print('vib_data_x is %d'%(vib_data_x[j]))
vib_data_x[j]=0
vib_data_y[j]=0
vib_data_z[j]=0
#z_2 = abs((z_tmp.copy())/20)
velocity_x = ((velocity_x_tmp)/size)
velocity_y = ((velocity_y_tmp)/size)
#z_tmp=0
velocity_x_tmp = 0
velocity_y_tmp = 0
#print ("ADXL345 on address 0x%x:" % (adxl345.address))
print('******')
print('the temperature are : ')
print('temp_1 = {0},temp_2={1}'.format(temperature[0],temperature[1]))
print('******')
print('the vibration is :')
print ("ADXL345 on address 0x%x:" % (0x00))
print("velocity_x = %.4fmm/s^2" % (velocity_x))
print("velocity_y = %.4fmm/s^2" % (velocity_y))
#print ("z = %.4fmm/s^2" % (z_2))
time.sleep(0.02)
print ("ADXL345 on address 0x%x:" % (0x01))
print("velocity_x_1 = %.4fmm/s^2" % (velocity_x_1))
print("velocity_y_1 = %.4fmm/s^2" % (velocity_y_1))
#print ("z_3 = %.4fmm/s^2" % (z_3))
#print(" velocity_z = %.4fmm/s" % (velocity_z))
time.sleep(0.05)
def thread_sensor2():
a=0
b=0
global x_3
global y_3
global z_3
global velocity_x_1
global velocity_y_1
global velocity_z_1
x_3=0
y_3=0
z_3=0
velocity_x_1= 0
velocity_y_1= 0
velocity_z_1= 0
velocity_x_tmp_1 = 0
velocity_y_tmp_1 = 0
velocity_z_tmp_1 = 0
vib_data_x_1 = np.zeros(size)
vib_data_y_1 = np.zeros(size)
adxl345_1 = ADXL345()
#adxl345_1.setoffset_y(0x06) # 0X01
while 1:
tca_select(1)
adxl345_1.setoffset_x(0x01)
adxl345_1.setoffset_y(0x03) # 0X01
#x_3,y_3,z_3=adxl345_1.read()
axes_1 = adxl345_1.getAxes_in_mm(True)
vib_data_y_1[a]= (axes_1['y'])
vib_data_x_1[a]= (axes_1['x'])
#z_3 = abs(axes_1['z'])
a+=1
if a==size:
a=0
for b in range(0,size):
velocity_x_tmp_1 = vib_data_x_1[b]+velocity_x_tmp_1
velocity_y_tmp_1 = vib_data_y_1[b]+velocity_y_tmp_1
vib_data_x_1[b]=0
vib_data_y_1[b]=0
velocity_x_2 = (copy.copy(velocity_x_tmp_1))/size
velocity_y_2 = (copy.copy(velocity_y_tmp_1))/size
velocity_x_1 = velocity_x_2
velocity_y_1 = velocity_y_2
velocity_x_tmp_1 = 0
velocity_y_tmp_1 = 0
#print("velocity_x_1 = %.4fmm/s" % (velocity_x_1))
#print("velocity_y_1 = %.4fmm/s" % (velocity_y_1))
#print ("z_3 = %.4fmm/s^2" % (z_3))
#print ("ADXL345_1 on address 0x%x:" % (adxl345_1.address))
#print('X={0}, Y={1}, Z={2}'.format(x_3, y_3, z_3))
time.sleep(0.05)
def thread_temperature():
while True:
for sensor in range(len(sensoid)):
temp_sensor = '/sys/bus/w1/devices/'+sensoid[sensor]+'/w1_slave'
temperature[sensor] = read_temp(temp_sensor)
#print('the %d temp sensor is '%(sensor+1),temperature[sensor])
my_thread1 = threading.Thread(target = thread_sensor1)
my_thread2 = threading.Thread(target = thread_sensor2)
#my_thread3 = threading.Thread(target = thread_mqtt)
#my_thread3 = threading.Thread(target = thread_print)
my_thread4 = threading.Thread(target = thread_temperature)
my_thread1.start()
my_thread2.start()
#my_thread3.start()
my_thread4.start()
print("this is main thread") # should not go here
```
## Complete the whole sensors:
```python=
from adxl345 import ADXL345
from Adafruit_GPIO import I2C
import time
import os
import threading
from scipy import pi
import matplotlib.pyplot as plt
import numpy as np
import copy
os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')
sensoid= ['3b-0000001817de','3b-0000001817d1']
global temp_sensor
global temperature
global x_1
global y_1
global z_1
global x_tmp
global size
size = 20
x_tmp = 0
temperature=[0,0]
def temp_raw(temp_sensor):
#self.temp_sensor=temp_sensor
f = open(temp_sensor,'r')
lines = f.readlines()
f.close()
return lines
def read_temp(temp_sensor):
lines = temp_raw(temp_sensor)
while lines[0].strip()[-3: ] != 'YES':
time.sleep(0.01)
lines = temp_raw()
temp_output = lines[1].find('t=')
if temp_output != -1:
temp_string = lines[1].strip()[temp_output+2:]
temp_c = float(temp_string)/1000.0
#temp_f = temp_c * 9.0/5.0+32.0
return temp_c
tca = I2C.get_i2c_device(address=0x70)
def tca_select(channel):
if channel > 7:
return
tca.writeRaw8(1 << channel)
def tca_set(mask):
if mask > 0xff:
return
tca.writeRaw8(mask)
def thread_sensor1():
i=0
k=0
global x_2
global velocity_x
global velocity_y
global velocity_z
global y_2
global z_2
x_2 = 0
y_2 = 0
z_2 = 0
velocity_x = 0
velocity_y = 0
velocity_z = 0
#y_tmp = 0
velocity_x_tmp = 0
velocity_y_tmp = 0
velocity_z_tmp = 0
#x_tmp = 0
z_tmp = 0
vib_data_x = np.zeros(size)
vib_data_y = np.zeros(size)
vib_data_z = np.zeros(size)
adxl345 = ADXL345()
adxl345.address = 0x53# 0XFC
adxl345.setRange(0x03)
print('test')
while 1:
tca_select(0)
adxl345.setoffset_x(0x01)
adxl345.setoffset_y(0x01) # 0X01
#x_1,y_1,z_1 = adxl345.read()
axes = adxl345.getAxes_in_mm(gforce = True)
z_1 = (axes['z'])
vib_data_y[i]= abs(axes['y'])
vib_data_x[i] = abs(axes['x']) # x acceleration
vib_data_z[i] = z_1
#print('vib_data_x_i is %d'%(vib_data_x[i]))
#print(z_1)
i+=1
if i==size:
i=0
for j in range(0,size):
velocity_x_tmp = vib_data_x[j]*0.05+velocity_x_tmp
velocity_y_tmp = vib_data_y[j]*0.05+velocity_y_tmp
#y_tmp = vib_data_y[j]+y_tmp
#z_tmp = vib_data_z[j]+z_tmp
#print('vib_data_x is %d'%(vib_data_x[j]))
vib_data_x[j]=0
vib_data_y[j]=0
vib_data_z[j]=0
#z_2 = abs((z_tmp.copy())/20)
velocity_x = ((velocity_x_tmp)/size)
velocity_y = ((velocity_y_tmp)/size)
#z_tmp=0
velocity_x_tmp = 0
velocity_y_tmp = 0
#print ("ADXL345 on address 0x%x:" % (adxl345.address))
print('******')
print('the temperature are : ')
print('temp_1 = {0},temp_2={1}'.format(temperature[0],temperature[1]))
print('******')
print('the vibration is :')
print ("ADXL345 on address 0x%x:" % (0x00))
print("velocity_x = %.4fmm/s" % (velocity_x))
print("velocity_y = %.4fmm/s" % (velocity_y))
#print ("z = %.4fmm/s^2" % (z_2))
time.sleep(0.02)
print ("ADXL345 on address 0x%x:" % (0x01))
print("velocity_x_1 = %.4fmm/s" % (velocity_x_1))
print("velocity_y_1 = %.4fmm/s" % (velocity_y_1))
#print ("z_3 = %.4fmm/s^2" % (z_3))
#print(" velocity_z = %.4fmm/s" % (velocity_z))
time.sleep(0.05)
def thread_sensor2():
a=0
b=0
global x_3
global y_3
global z_3
global velocity_x_1
global velocity_y_1
global velocity_z_1
x_3=0
y_3=0
z_3=0
velocity_x_1= 0
velocity_y_1= 0
velocity_z_1= 0
velocity_x_tmp_1 = 0
velocity_y_tmp_1 = 0
velocity_z_tmp_1 = 0
vib_data_x_1 = np.zeros(size)
vib_data_y_1 = np.zeros(size)
adxl345_1 = ADXL345()
#adxl345_1.setoffset_y(0x06) # 0X01
while 1:
tca_select(1)
adxl345_1.setoffset_x(0x01)
adxl345_1.setoffset_y(0x03) # 0X01
#x_3,y_3,z_3=adxl345_1.read()
axes_1 = adxl345_1.getAxes_in_mm(True)
vib_data_y_1[a]= abs(axes_1['y'])
vib_data_x_1[a]= abs(axes_1['x'])
#z_3 = abs(axes_1['z'])
a+=1
if a==size:
a=0
for b in range(0,size):
velocity_x_tmp_1 = vib_data_x_1[b]*0.05+velocity_x_tmp_1
velocity_y_tmp_1 = vib_data_y_1[b]*0.05+velocity_y_tmp_1
vib_data_x_1[b]=0
vib_data_y_1[b]=0
velocity_x_2 = (copy.copy(velocity_x_tmp_1))/size
velocity_y_2 = (copy.copy(velocity_y_tmp_1))/size
velocity_x_1 = velocity_x_2
velocity_y_1 = velocity_y_2
velocity_x_tmp_1 = 0
velocity_y_tmp_1 = 0
#print("velocity_x_1 = %.4fmm/s" % (velocity_x_1))
#print("velocity_y_1 = %.4fmm/s" % (velocity_y_1))
#print ("z_3 = %.4fmm/s^2" % (z_3))
#print ("ADXL345_1 on address 0x%x:" % (adxl345_1.address))
#print('X={0}, Y={1}, Z={2}'.format(x_3, y_3, z_3))
time.sleep(0.05)
def thread_temperature():
while True:
for sensor in range(len(sensoid)):
temp_sensor = '/sys/bus/w1/devices/'+sensoid[sensor]+'/w1_slave'
temperature[sensor] = read_temp(temp_sensor)
#print('the %d temp sensor is '%(sensor+1),temperature[sensor])
my_thread1 = threading.Thread(target = thread_sensor1)
my_thread2 = threading.Thread(target = thread_sensor2)
#my_thread3 = threading.Thread(target = thread_mqtt)
#my_thread3 = threading.Thread(target = thread_print)
my_thread4 = threading.Thread(target = thread_temperature)
my_thread1.start()
my_thread2.start()
#my_thread3.start()
my_thread4.start()
print("this is main thread") # should not go here
```
# LOCK with tca:
```python=
from adxl345 import ADXL345
import time
import os
import paho.mqtt.client as mqtt
import threading
from scipy import pi
import matplotlib.pyplot as plt
from Adafruit_GPIO import I2C
import numpy as np
import copy
tca = I2C.get_i2c_device(address=0x70)
lock_tca = threading.Lock()
def tca_select(channel):
if channel > 7:
return
tca.writeRaw8(1 << channel)
def tca_set(mask):
if mask > 0xff:
return
tca.writeRaw8(mask)
def thread_sensor1( lock_tca):
global velocity_x
global velocity_y
global z_1
global size
size = 40
velocity_x = 0
velocity_y = 0
z_1 = 0
i=0
velocity_x_tmp = 0
velocity_y_tmp = 0
vib_data_x = np.zeros(size)
vib_data_y = np.zeros(size)
tca_select(0)
adxl345 = ADXL345()
while True:
lock_tca.acquire()
tca_select(0)
adxl345.setoffset_x(0x00)
adxl345.setoffset_y(0xFE)
axes =adxl345.getAxes_in_mm(gforce = True)
#x_1,y_1,z_1 = adxl345.read()
#x_1 = axes['x']
#y_1 = axes['y']
vib_data_x[i] = (axes['x'])
vib_data_y[i]= (axes['y'])
lock_tca.release()
i+=1
if i==size:
i=0
for j in range(0,size):
velocity_x_tmp = vib_data_x[j]+velocity_x_tmp
velocity_y_tmp = vib_data_y[j]+velocity_y_tmp
vib_data_x[j]=0
vib_data_y[j]=0
velocity_x = ((velocity_x_tmp)/size)
velocity_y = ((velocity_y_tmp)/size)
velocity_x_tmp = 0
velocity_y_tmp = 0
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print (" x = %.4fmm/s^2" % ( velocity_x))
print (" y = %.4fmm/s^2" % ( velocity_y ))
print ("ADXL345 on address 0x%x:" % (0x01))
print("velocity_x_1 = %.4fmm/s^2" % (velocity_x_1))
print("velocity_y_1 = %.4fmm/s^2" % (velocity_y_1))
time.sleep(0.01)
def thread_sensor2( lock_tca):
a=0
global velocity_x_1
global velocity_y_1
global size_1
size_1 = 40
velocity_x_1= 0
velocity_y_1= 0
velocity_x_tmp_1 = 0
velocity_y_tmp_1 = 0
vib_data_x_1 = np.zeros(size_1)
vib_data_y_1 = np.zeros(size_1)
tca_select(1)
adxl345_1 = ADXL345()
while True:
lock_tca.acquire()
tca_select(1)
adxl345_1.setoffset_x(0xFE)
adxl345_1.setoffset_y(0x02)
axes_1 = adxl345_1.getAxes_in_mm(gforce = True)
vib_data_x_1[a]= (axes_1['x'])
vib_data_y_1[a]= (axes_1['y'])
lock_tca.release()
a+=1
if a==size_1:
a=0
for b in range(0,size_1):
velocity_x_tmp_1 = vib_data_x_1[b]+velocity_x_tmp_1
velocity_y_tmp_1 = vib_data_y_1[b]+velocity_y_tmp_1
vib_data_x_1[b]=0
vib_data_y_1[b]=0
velocity_x_1 = ((velocity_x_tmp_1)/size_1)
velocity_y_1 = ((velocity_y_tmp_1)/size_1)
velocity_x_tmp_1 = 0
velocity_y_tmp_1 = 0
time.sleep(0.01)
my_thread1 = threading.Thread(target = thread_sensor1, args=(lock_tca,))
my_thread2 = threading.Thread(target = thread_sensor2, args=(lock_tca,))
my_thread1.start()
my_thread2.start()
print("this is main thread") # should not go here
```
# With one thread:
```python=
from adxl345 import ADXL345
import time
import os
import paho.mqtt.client as mqtt
import threading
from scipy import pi
import matplotlib.pyplot as plt
from Adafruit_GPIO import I2C
import numpy as np
import copy
tca = I2C.get_i2c_device(address=0x70)
def tca_select(channel):
if channel > 7:
return
tca.writeRaw8(1 << channel)
def tca_set(mask):
if mask > 0xff:
return
tca.writeRaw8(mask)
def thread_sensor1():
global velocity_x
global velocity_y
global velocity_x_1
global velocity_y_1
global z_1
global size
size = 20
velocity_x = 0
velocity_y = 0
velocity_x_1= 0
velocity_y_1= 0
z_1 = 0
i=0
a=0
velocity_x_tmp = 0
velocity_y_tmp = 0
velocity_x_tmp_1 = 0
velocity_y_tmp_1 = 0
vib_data_x = np.zeros(size)
vib_data_y = np.zeros(size)
vib_data_x_1 = np.zeros(size)
vib_data_y_1 = np.zeros(size)
#tca_select(0)
adxl345 = ADXL345()
adxl345_1 = ADXL345()
while True:
tca_select(0)
adxl345.setoffset_x(0x00)
adxl345.setoffset_y(0xFE)
axes =adxl345.getAxes_in_mm(gforce = True)
#x_1,y_1,z_1 = adxl345.read()
#x_1 = axes['x']
#y_1 = axes['y']
vib_data_x[i] = (axes['x'])
vib_data_y[i]= (axes['y'])
i+=1
if i==size:
i=0
for j in range(0,size):
velocity_x_tmp = vib_data_x[j]+velocity_x_tmp
velocity_y_tmp = vib_data_y[j]+velocity_y_tmp
vib_data_x[j]=0
vib_data_y[j]=0
velocity_x = ((velocity_x_tmp)/size)
velocity_y = ((velocity_y_tmp)/size)
velocity_x_tmp = 0
velocity_y_tmp = 0
tca_select(1)
adxl345_1.setoffset_x(0xFF)
adxl345_1.setoffset_y(0x02)
axes_1 = adxl345_1.getAxes_in_mm(gforce = True)
vib_data_x_1[a]= (axes_1['x'])
vib_data_y_1[a]= (axes_1['y'])
a+=1
if a==size:
a=0
for b in range(0,size):
velocity_x_tmp_1 = vib_data_x_1[b]+velocity_x_tmp_1
velocity_y_tmp_1 = vib_data_y_1[b]+velocity_y_tmp_1
vib_data_x_1[b]=0
vib_data_y_1[b]=0
velocity_x_1 = ((velocity_x_tmp_1)/size)
velocity_y_1 = ((velocity_y_tmp_1)/size)
velocity_x_tmp_1 = 0
velocity_y_tmp_1 = 0
print ("ADXL345 on address 0x%x:" % (adxl345.address))
print (" x = %.4fmm/s^2" % ( velocity_x))
print (" y = %.4fmm/s^2" % ( velocity_y ))
print ("ADXL345 on address 0x%x:" % (0x01))
print("velocity_x_1 = %.4fmm/s^2" % (velocity_x_1))
print("velocity_y_1 = %.4fmm/s^2" % (velocity_y_1))
time.sleep(0.01)
my_thread1 = threading.Thread(target = thread_sensor1)
#my_thread2 = threading.Thread(target = thread_sensor2)
#my_thread3 = threading
my_thread1.start()
#my_thread2.start()
print("this is main thread") # should not go here
```
Sign in with Wallet
Connect another wallet