Scribd
Upload
11 views20 pages

Python Scripts for Sensor Data Handling

The document consists of multiple Python scripts that interact with hardware components for monitoring and controlling environmental parameters. The scripts include functionalities for reading temperature, controlling actuators and motors, monitoring pH and dissolved oxygen levels, and sending alerts via SMS based on sensor readings. Each script is designed to run continuously, performing specific tasks related to data collection and hardware control in an automated system.

Uploaded by

Smiley
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
11 views20 pages

Python Scripts for Sensor Data Handling

The document consists of multiple Python scripts that interact with hardware components for monitoring and controlling environmental parameters. The scripts include functionalities for reading temperature, controlling actuators and motors, monitoring pH and dissolved oxygen levels, and sending alerts via SMS based on sensor readings. Each script is designed to run continuously, performing specific tasks related to data collection and hardware control in an automated system.

Uploaded by

Smiley
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

Temp.

py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import glob
import time
import requests

# ==== CONFIG ====


PHP_URL = "[Link] # your PHP endpoint

# Load 1-Wire kernel modules


[Link]('modprobe w1-gpio')
[Link]('modprobe w1-therm')

# Find DS18B20 device folder


base_dir = '/sys/bus/w1/devices/'
device_folder = [Link](base_dir + '28-*')[0] # sensor address starts with 28-
device_file = device_folder + '/w1_slave'

def read_temp_raw():
"""Read raw lines from sensor file"""
with open(device_file, 'r') as f:
return [Link]()

def read_temperature():
"""Parse temperature from raw sensor data"""
lines = read_temp_raw()
# Wait until the reading is complete
while lines[0].strip()[-3:] != 'YES':
[Link](0.2)
lines = read_temp_raw()

equals_pos = lines[1].find('t=')
if equals_pos != -1:
temp_string = lines[1][equals_pos+2:]
temp_c = float(temp_string) / 1000.0
return round(temp_c, 2)
return None

def send_to_php(temp):
"""Send temperature data to PHP"""
try:
params = {'temp': temp}
r = [Link](PHP_URL, params=params, timeout=5)
if r.status_code == 200:
print(f"Sent: Temp={temp} C | Response OK")
else:
print(f"Server response: {r.status_code}")
except Exception as e:
print("Error sending data:", e)

if __name__ == "__main__":
while True:
temp_value = read_temperature()
if temp_value is not None:
print(f"Temperature: {temp_value} C")
send_to_php(temp_value)
else:
print("Sensor read failed.")
[Link](10) # adjust interval

[Link]
# -*- coding: ascii -*-
import lgpio
import time
import threading

# ===============================
# LINEAR ACTUATOR + SSR
# ===============================
IN1 = 25
IN2 = 8
SSR_PIN = 26

# ===============================
# STEPPER MOTOR
# ===============================
M_IN1 = 21
M_IN2 = 20
M_IN3 = 16
M_IN4 = 12
STEP_SEQUENCE = [
[1, 0, 0, 0],
[1, 1, 0, 0],
[0, 1, 0, 0],
[0, 1, 1, 0],
[0, 0, 1, 0],
[0, 0, 1, 1],
[0, 0, 0, 1],
[1, 0, 0, 1],
]

STEP_ANGLE = 1.8
MICROSTEP_FACTOR = 2

# ===============================
# DIGITAL POT X9C103S
# ===============================
INC = 9
UD = 10
CS = 11
MAX_STEPS = 99
half_pos = MAX_STEPS // 2

# ===============================
# SETUP
# ===============================
chip = 0
h = lgpio.gpiochip_open(chip)
for pin in [IN1, IN2, SSR_PIN, M_IN1, M_IN2, M_IN3, M_IN4, INC, UD, CS]:
lgpio.gpio_claim_output(h, pin)
lgpio.gpio_write(h, pin, 0)

# ===============================
# DIGITAL POT FUNCTIONS
# ===============================
def pulse_inc():
lgpio.gpio_write(h, INC, 1)
[Link](0.001)
lgpio.gpio_write(h, INC, 0)
[Link](0.001)

def set_to_max():
lgpio.gpio_write(h, CS, 0)
lgpio.gpio_write(h, UD, 1)
for _ in range(MAX_STEPS):
pulse_inc()
lgpio.gpio_write(h, CS, 1)

def set_to_position(position):
if position < 0:
position = 0
if position > MAX_STEPS:
position = MAX_STEPS

steps_down = MAX_STEPS - position


lgpio.gpio_write(h, CS, 0)
lgpio.gpio_write(h, UD, 0)
for _ in range(steps_down):
pulse_inc()
lgpio.gpio_write(h, CS, 1)

def pot_loop():
while True:
print("Pot to MAX for 10 seconds")
set_to_max()
[Link](10)

print("Pot to HALF for 5 seconds")


set_to_position(half_pos)
[Link](5)

# ===============================
# ACTUATOR FUNCTIONS
# ===============================
def extend():
print("Extending actuator")
lgpio.gpio_write(h, IN1, 1)
lgpio.gpio_write(h, IN2, 0)

def retract():
print("Retracting actuator")
lgpio.gpio_write(h, IN1, 0)
lgpio.gpio_write(h, IN2, 1)

def stop_actuator():
print("Stopping actuator")
lgpio.gpio_write(h, IN1, 0)
lgpio.gpio_write(h, IN2, 0)

def trigger_ssr(state):
lgpio.gpio_write(h, SSR_PIN, 1 if state else 0)
print("SSR ON" if state else "SSR OFF")

# ===============================
# STEPPER FUNCTIONS
# ===============================
def angle_to_steps(angle):
steps_per_rev = int(360 / STEP_ANGLE * MICROSTEP_FACTOR)
steps = int(steps_per_rev * (angle / 360))
return steps

def rotate_steps(pins, steps, delay=0.002):


for step_count in range(steps):
seq = STEP_SEQUENCE[step_count % len(STEP_SEQUENCE)]
for pin, val in zip(pins, seq):
lgpio.gpio_write(h, pin, val)
[Link](delay)
for pin in pins:
lgpio.gpio_write(h, pin, 0)
def rotate_reverse(pins, steps, delay=0.002):
for step_count in range(steps):
seq = STEP_SEQUENCE[-(step_count % len(STEP_SEQUENCE)) - 1]
for pin, val in zip(pins, seq):
lgpio.gpio_write(h, pin, val)
[Link](delay)
for pin in pins:
lgpio.gpio_write(h, pin, 0)

def stepper_sequence():
pins = [M_IN1, M_IN2, M_IN3, M_IN4]
angles = [25, 45, 90, 135, 155]
delay_between_angles = 5
current_angle = 25

for target_angle in angles:


steps = angle_to_steps(abs(target_angle - current_angle))
if target_angle > current_angle:
rotate_steps(pins, steps)
else:
rotate_reverse(pins, steps)
print("Stepper moved")
current_angle = target_angle
[Link](delay_between_angles)

steps = angle_to_steps(abs(25 - current_angle))


if 25 < current_angle:
rotate_reverse(pins, steps)
else:
rotate_steps(pins, steps)
print("Stepper reset")
[Link](delay_between_angles)

# ===============================
# MAIN SEQUENCE
# ===============================
try:
print("Triggering SSR")
trigger_ssr(True)
[Link](5)

# Start potentiometer loop in background


pot_thread = [Link](target=pot_loop)
pot_thread.start()

extend()
[Link](10)
stop_actuator()
print("Actuator extended")

stepper_start_time = [Link]()
retract_triggered = False
total_runtime = 0

while total_runtime < 100:


elapsed = [Link]() - stepper_start_time
total_runtime = elapsed

if elapsed >= 40 and not retract_triggered:


print("40 seconds elapsed, retracting actuator")
retract()
[Link](10)
stop_actuator()
print("Actuator retracted")
retract_triggered = True

if int(elapsed) % 10 == 0:
print("Stepper run")
stepper_sequence()

[Link](1)

print("Waiting 60 seconds after retraction")


[Link](60)
trigger_ssr(False)
print("SSR OFF. End.")

except KeyboardInterrupt:
print("Interrupted")

finally:
stop_actuator()
trigger_ssr(False)
lgpio.gpiochip_close(h)
print("GPIO closed")

[Link]
import requests
import json
import os
import serial
import time
import lgpio

# GPIO pins
TRIG = 13
ECHO = 19
CHIP = 0

# Feed distance threshold (2 feet approx 60 cm)


FEED_LOW_CM = 60.96

# Endpoint
DATA_URL = "[Link]
ALERT_STATE_FILE = "/home/pi/alert_state.json"

# Thresholds for sensors


TEMP_MIN, TEMP_MAX = 25, 31
PH_MIN, PH_MAX = 7, 8.5
DO_MIN, DO_MAX = 0.1, 1

# SIM900A Serial config


SERIAL_PORT = "/dev/serial0"
BAUD_RATE = 9600
ALERT_PHONE = "+639062260719"

# Setup GPIO chip


h = lgpio.gpiochip_open(CHIP)
lgpio.gpio_claim_output(h, TRIG)
lgpio.gpio_claim_input(h, ECHO)

def get_distance():
lgpio.gpio_write(h, TRIG, 0)
[Link](0.02)

lgpio.gpio_write(h, TRIG, 1)
[Link](0.00001)
lgpio.gpio_write(h, TRIG, 0)

start_time = [Link]()
while lgpio.gpio_read(h, ECHO) == 0:
start_time = [Link]()

end_time = [Link]()
while lgpio.gpio_read(h, ECHO) == 1:
end_time = [Link]()

duration = end_time - start_time


distance = (duration * 34300) / 2
return distance
def send_sms(message):
try:
ser = [Link](SERIAL_PORT, BAUD_RATE, timeout=5)
[Link](2)
[Link](b"AT\r")
[Link](2)
[Link](b"AT+CMGF=1\r")
[Link](2)
[Link](f'AT+CMGS="{ALERT_PHONE}"\r'.encode())
[Link](2)
[Link](f"{message}\x1A".encode())
[Link](3)
[Link]()
print("[SMS SENT]", message)
except Exception as e:
print("[ERROR] Failed to send SMS:", e)

def load_alert_state():
if not [Link](ALERT_STATE_FILE):
init_state = {
"temp": {"alert": False, "last_alert": 0},
"ph": {"alert": False, "last_alert": 0},
"do": {"alert": False, "last_alert": 0},
"feeds": {"alert": False, "last_alert": 0}
}
with open(ALERT_STATE_FILE, "w") as f:
[Link](init_state, f)
with open(ALERT_STATE_FILE, "r") as f:
return [Link](f)

def save_alert_state(state):
with open(ALERT_STATE_FILE, "w") as f:
[Link](state, f)

while True:
try:
# Read sensor data from server
response = [Link](DATA_URL, timeout=5)
data = [Link]()

temp = float([Link]("temp") or 0)
ph = float([Link]("ph") or 0)
do = float([Link]("do") or 0)

# Read feed distance


distance = get_distance()

print("Temp:", temp, "pH:", ph, "DO:", do, "Distance:", distance, "cm")

state = load_alert_state()
current_time = [Link]()

# Temperature alert
if temp < TEMP_MIN or temp > TEMP_MAX:
if not state["temp"]["alert"] or current_time - state["temp"]["last_alert"] > 300:
send_sms(f"ALERT: Temperature out of range ({temp:.2f} C)!")
state["temp"]["alert"] = True
state["temp"]["last_alert"] = current_time
else:
if state["temp"]["alert"]:
send_sms(f"Recovery: Temperature normal ({temp:.2f} C)")
state["temp"]["alert"] = False
state["temp"]["last_alert"] = current_time

# pH alert
if ph < PH_MIN or ph > PH_MAX:
if not state["ph"]["alert"] or current_time - state["ph"]["last_alert"] > 300:
send_sms(f"ALERT: pH out of range ({ph:.2f})!")
state["ph"]["alert"] = True
state["ph"]["last_alert"] = current_time
else:
if state["ph"]["alert"]:
send_sms(f"Recovery: pH normal ({ph:.2f})")
state["ph"]["alert"] = False
state["ph"]["last_alert"] = current_time

# Dissolved Oxygen (DO) alert


if do < DO_MIN or do > DO_MAX:
if not state["do"]["alert"] or current_time - state["do"]["last_alert"] > 300:
send_sms(f"ALERT: Dissolved Oxygen out of range ({do:.2f} mg/L)!")
state["do"]["alert"] = True
state["do"]["last_alert"] = current_time
else:
if state["do"]["alert"]:
send_sms(f"Recovery: Dissolved Oxygen normal ({do:.2f} mg/L)")
state["do"]["alert"] = False
state["do"]["last_alert"] = current_time

# Feed level alert


if distance > FEED_LOW_CM:
if not state["feeds"]["alert"] or current_time - state["feeds"]["last_alert"] > 900:
send_sms(f"ALERT: Feed level low distance {distance:.2f} cm (approx 10kg left)")
state["feeds"]["alert"] = True
state["feeds"]["last_alert"] = current_time
else:
if state["feeds"]["alert"]:
send_sms(f"Feed level normal distance {distance:.2f} cm")
state["feeds"]["alert"] = False
state["feeds"]["last_alert"] = current_time

save_alert_state(state)

except Exception as e:
print("[ERROR]", e)

[Link](10)
[Link]
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import time
import board
import busio
import requests
from adafruit_ads1x15.analog_in import AnalogIn
import adafruit_ads1x15.ads1115 as ADS

# =============================
# CONFIGURATION
# =============================

# URL of your PHP file (change IP if PHP is hosted elsewhere)


PHP_URL = "[Link]

# ADS1115 gain setting (adjust depending on sensor output range)


ADS_GAIN = 1

# Calibration constants for pH sensor (adjust after calibration)


PH_SLOPE = -2.66 # Example slope, calibrate properly
PH_OFFSET = 21.34 # Example offset

# Calibration constants for DO sensor (adjust after calibration)


DO_ZERO = 0.0 # Voltage at 0 mg/L
DO_MAX_VOLTAGE = 2.0 # Voltage at max DO
DO_MAX_VALUE = 14.0 # mg/L at max DO

# =============================
# INITIALIZE ADC
# =============================

i2c = busio.I2C([Link], [Link])


ads = ADS.ADS1115(i2c)
[Link] = ADS_GAIN

# Assign ADS1115 channels


ph_channel = AnalogIn(ads, ADS.P0) # pH sensor on A0
do_channel = AnalogIn(ads, ADS.P1) # DO sensor on A1

# =============================
# FUNCTIONS
# =============================

def read_ph(voltage):
"""Convert voltage to pH based on calibration."""
return (voltage - PH_OFFSET) / PH_SLOPE

def read_do(voltage):
"""Convert voltage to DO mg/L based on calibration."""
return (voltage - DO_ZERO) * (DO_MAX_VALUE / DO_MAX_VOLTAGE)

def send_to_server(ph, do):


"""Send readings to PHP server (ph + do only)."""
try:
params = {
"ph": f"{ph:.2f}",
"do": f"{do:.2f}"
}
r = [Link](PHP_URL, params=params, timeout=5)
if r.status_code == 200:
print("[?] Data sent successfully to server.")
else:
print(f"[??] Failed to send data. HTTP {r.status_code}")
except Exception as e:
print(f"[?] Error sending data: {e}")

# =============================
# MAIN LOOP
# =============================

try:
while True:
# Read voltages
ph_voltage = ph_channel.voltage
do_voltage = do_channel.voltage

# Convert to values
ph_value = read_ph(ph_voltage)
do_value = read_do(do_voltage)

# Print readings
print(f"pH: {ph_value:.2f} | DO: {do_value:.2f} mg/L")

# Send to PHP server


send_to_server(ph_value, do_value)

# Wait before next reading


[Link](5)

except KeyboardInterrupt:
print("\n[??] Program stopped by user.")

You might also like