102 lines
2.8 KiB
Python
Executable File
102 lines
2.8 KiB
Python
Executable File
import serial
|
|
import json
|
|
import time
|
|
|
|
# Initialize serial ports for the three sensors
|
|
ser3 = serial.Serial(
|
|
port='/dev/ttyAMA3',
|
|
baudrate=115200,
|
|
parity=serial.PARITY_EVEN,
|
|
stopbits=serial.STOPBITS_ONE,
|
|
bytesize=serial.EIGHTBITS,
|
|
timeout = 1
|
|
)
|
|
|
|
ser4 = serial.Serial(
|
|
port='/dev/ttyAMA4',
|
|
baudrate=115200,
|
|
parity=serial.PARITY_EVEN,
|
|
stopbits=serial.STOPBITS_ONE,
|
|
bytesize=serial.EIGHTBITS,
|
|
timeout = 1
|
|
)
|
|
|
|
ser5 = serial.Serial(
|
|
port='/dev/ttyAMA5',
|
|
baudrate=115200,
|
|
parity=serial.PARITY_EVEN,
|
|
stopbits=serial.STOPBITS_ONE,
|
|
bytesize=serial.EIGHTBITS,
|
|
timeout = 1
|
|
)
|
|
|
|
# Function to read and parse sensor data
|
|
def read_sensor_data(ser, sonde_id):
|
|
try:
|
|
# Send the command to request data (e.g., data for 60 seconds)
|
|
ser.write(b'\x81\x12\x6D')
|
|
|
|
# Read the response
|
|
byte_data = ser.readline()
|
|
|
|
# Extract the state byte and PM data from the response
|
|
state_byte = int.from_bytes(byte_data[2:3], byteorder='big')
|
|
state_bits = [int(bit) for bit in bin(state_byte)[2:].zfill(8)]
|
|
|
|
PM1 = int.from_bytes(byte_data[9:11], byteorder='big') / 10
|
|
PM25 = int.from_bytes(byte_data[11:13], byteorder='big') / 10
|
|
PM10 = int.from_bytes(byte_data[13:15], byteorder='big') / 10
|
|
|
|
# Create a dictionary with the parsed data
|
|
data = {
|
|
'sondeID': sonde_id,
|
|
'PM1': PM1,
|
|
'PM25': PM25,
|
|
'PM10': PM10
|
|
}
|
|
|
|
return data
|
|
except Exception as e:
|
|
print(f"Error reading from sensor {sonde_id}: {e}")
|
|
return None
|
|
|
|
# Function to create a JSON object with all sensor data
|
|
def collect_all_sensor_data():
|
|
all_data = {}
|
|
|
|
# Read data from each sensor and add to the all_data dictionary
|
|
sensor_data_3 = read_sensor_data(ser3, 'USB2')
|
|
sensor_data_4 = read_sensor_data(ser4, 'USB3')
|
|
sensor_data_5 = read_sensor_data(ser5, 'USB4')
|
|
|
|
# Store the data for each sensor in the dictionary
|
|
if sensor_data_3:
|
|
all_data['sensor_3'] = sensor_data_3
|
|
if sensor_data_4:
|
|
all_data['sensor_4'] = sensor_data_4
|
|
if sensor_data_5:
|
|
all_data['sensor_5'] = sensor_data_5
|
|
|
|
return all_data
|
|
|
|
# Main script to run once
|
|
if __name__ == "__main__":
|
|
try:
|
|
# Collect data from all sensors
|
|
data = collect_all_sensor_data()
|
|
|
|
# Convert data to JSON
|
|
json_data = json.dumps(data, indent=4)
|
|
|
|
# Define the output file path
|
|
output_file = "/var/www/nebuleair_pro_4g/loop/data.json" # Change this to your desired file path
|
|
|
|
# Write the JSON data to the file
|
|
with open(output_file, 'w') as file:
|
|
file.write(json_data)
|
|
|
|
print(f"Data successfully written to {output_file}")
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|