How to Use Python to Control Real-World Hardware: Sensors, Motors, and IoT Devices

import RPi.GPIO as GPIO
import time
from typing import List, Optional

class GPIOManager:
    def __init__(self, mode=GPIO.BCM):
        GPIO.setmode(mode)
        GPIO.setwarnings(False)
        self.pin_modes = {}

    def setup_pin(self, pin: int, mode: str, pull_up_down: Optional[str] = None):
        """Set a pin as either input or output."""
        if mode.upper() == 'OUT':
            GPIO.setup(pin, GPIO.OUT)
            self.pin_modes[pin] = 'OUT'
        elif mode.upper() == 'IN':
            if pull_up_down:
                if pull_up_down.upper() == 'UP':
                    GPIO.setup(pin, GPIO.IN, pull_up_down=GPIO.PUD_UP)
                elif pull_up_down.upper() == 'DOWN':
                    GPIO.setup(pin, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
            else:
                GPIO.setup(pin, GPIO.IN)
            self.pin_modes[pin] = 'IN'
        else:
            raise ValueError(f"Mode must be 'IN' or 'OUT'. Got: {mode}")

    def write_pin(self, pin: int, state: bool):
        """Send a HIGH (True) or LOW (False) signal to an output pin."""
        if self.pin_modes.get(pin) != 'OUT':
            raise ValueError(f"Pin {pin} is not an output.")
        GPIO.output(pin, GPIO.HIGH if state else GPIO.LOW)

    def read_pin(self, pin: int) -> bool:
        """Check if an input pin is receiving a HIGH (True) or LOW (False) signal."""
        if self.pin_modes.get(pin) != 'IN':
            raise ValueError(f"Pin {pin} is not an input.")
        return GPIO.input(pin) == GPIO.HIGH

    def pwm_control(self, pin: int, frequency: float, duty_cycle: float):
        """Create a Pulse Width Modulation signal. Great for dimming LEDs or controlling motor speed."""
        self.setup_pin(pin, 'OUT')
        pwm = GPIO.PWM(pin, frequency)
        pwm.start(duty_cycle)
        return pwm

    def monitor_button(self, pin: int, callback, bounce_time: int = 200):
        """Watch a button pin. The 'bounce_time' stops false triggers from physical switch jitter."""
        self.setup_pin(pin, 'IN', pull_up_down='UP')
        GPIO.add_event_detect(pin, GPIO.FALLING, callback=callback, bouncetime=bounce_time)

    def cleanup(self):
        """Always run this when your program ends to reset the pins safely."""
        GPIO.cleanup()

# Here's how I would use it in a simple project:
try:
    gpio = GPIOManager()

    # Setup an LED on pin 17
    led_pin = 17
    gpio.setup_pin(led_pin, 'OUT')

    print("Blinking the LED 5 times...")
    for i in range(5):
        gpio.write_pin(led_pin, True)  # Turn on
        time.sleep(0.5)
        gpio.write_pin(led_pin, False) # Turn off
        time.sleep(0.5)

    # Use PWM to fade an LED
    print("Now fading an LED with PWM...")
    pwm = gpio.pwm_control(18, 100, 50)  # Pin 18, 100Hz, start at 50% brightness
    time.sleep(2)
    pwm.ChangeDutyCycle(25)  # Dim to 25%
    time.sleep(2)
    pwm.stop()

finally:
    gpio.cleanup()  # This is crucial!

import smbus2
import struct
import time

class I2CManager:
    def __init__(self, bus_number: int = 1):
        self.bus = smbus2.SMBus(bus_number)

    def scan_devices(self):
        """Find all I2C devices connected to the bus."""
        found = []
        for address in range(0x03, 0x78):
            try:
                self.bus.read_byte(address)
                found.append(hex(address))
            except:
                pass
        return found

    def read_register(self, address: int, register: int) -> int:
        """Read one byte of data from a specific register on a device."""
        return self.bus.read_byte_data(address, register)

    def write_register(self, address: int, register: int, value: int):
        """Write one byte of data to a specific register."""
        self.bus.write_byte_data(address, register, value)

class BMP280_Sensor:
    """A driver for the BMP280 temperature and barometric pressure sensor."""

    def __init__(self, i2c_bus: I2CManager, address: int = 0x76):
        self.i2c = i2c_bus
        self.address = address
        self.calibration = {}
        self._setup_sensor()

    def _setup_sensor(self):
        """Read the unique calibration data from the chip and configure it."""
        # First, check we're talking to the right chip
        chip_id = self.i2c.read_register(self.address, 0xD0)
        if chip_id != 0x58:
            raise RuntimeError(f"Wrong chip. Expected BMP280 (ID 0x58), got {hex(chip_id)}")

        # The chip stores unique calibration numbers. We must read them.
        cal_data = self.i2c.read_block(self.address, 0x88, 24)

        # Convert bytes into usable numbers (this is specific to the BMP280)
        self.calibration['dig_T1'] = cal_data[0] | (cal_data[1] << 8)
        self.calibration['dig_T2'] = self._to_signed(cal_data[2] | (cal_data[3] << 8))
        self.calibration['dig_T3'] = self._to_signed(cal_data[4] | (cal_data[5] << 8))
        # ... (similar for pressure calibration values P1-P9)

        # Tell the sensor to start taking normal measurements
        self.i2c.write_register(self.address, 0xF4, 0x27)
        time.sleep(0.005)

    def _to_signed(self, value: int) -> int:
        """Helper to convert raw bytes to a signed integer."""
        if value & 0x8000:
            return value - 0x10000
        return value

    def read(self) -> dict:
        """Trigger a measurement and return calculated temperature and pressure."""
        # Read 6 bytes of raw data
        data = self.i2c.read_block(self.address, 0xF7, 6)
        temp_raw = (data[3] << 12) | (data[4] << 4) | (data[5] >> 4)
        press_raw = (data[0] << 12) | (data[1] << 4) | (data[2] >> 4)

        # Use the calibration data to convert raw numbers to real values
        temperature = self._calculate_temperature(temp_raw)
        pressure = self._calculate_pressure(press_raw, temperature)

        return {'temperature_c': temperature, 'pressure_hPa': pressure}

    def _calculate_temperature(self, raw_temp: int) -> float:
        """Apply the calibration formula from the BMP280 datasheet."""
        dig = self.calibration
        var1 = ((raw_temp / 16384.0) - (dig['dig_T1'] / 1024.0)) * dig['dig_T2']
        var2 = ((raw_temp / 131072.0) - (dig['dig_T1'] / 8192.0)) * dig['dig_T3']
        t_fine = var1 + var2
        return t_fine / 5120.0

    def _calculate_pressure(self, raw_press: int, t_fine: float) -> float:
        """Apply the more complex pressure calibration formula."""
        dig = self.calibration
        # ... (detailed compensation algorithm)
        return pressure_value  # in hPa

# Using the sensor
print("I2C Sensor Example")
print("=" * 40)
try:
    i2c = I2CManager(1)
    print(f"Devices on bus: {i2c.scan_devices()}")

    sensor = BMP280_Sensor(i2c)
    reading = sensor.read()
    print(f"Temperature: {reading['temperature_c']:.2f} °C")
    print(f"Pressure: {reading['pressure_hPa']:.2f} hPa")

except Exception as e:
    print(f"Error: {e}")

import serial
import serial.tools.list_ports
import threading
import time

class SerialMonitor:
    def __init__(self):
        self.connection = None
        self.monitor_thread = None
        self.stop_signal = threading.Event()

    def find_ports(self):
        """Show all serial ports (like USB adapters) available on the computer."""
        return [p.device for p in serial.tools.list_ports.comports()]

    def connect(self, port_name: str, baud_rate: int = 9600):
        """Open a connection to a serial device."""
        self.connection = serial.Serial(
            port=port_name,
            baudrate=baud_rate,
            timeout=2.0
        )
        print(f"Connected to {port_name} at {baud_rate} baud.")

    def start_listening(self, data_handler):
        """Start a background thread that constantly listens for incoming data."""
        if not self.connection:
            raise RuntimeError("Not connected to a port.")

        self.stop_signal.clear()

        def listen_loop():
            buffer = ""
            while not self.stop_signal.is_set():
                if self.connection.in_waiting:
                    # Read bytes and decode to text
                    text = self.connection.read(self.connection.in_waiting).decode('ascii', errors='ignore')
                    buffer += text
                    # Split by newline to get complete sentences
                    while '\\n' in buffer:
                        line, buffer = buffer.split('\\n', 1)
                        data_handler(line.strip())  # Send the line to our handler function
                time.sleep(0.01)

        self.monitor_thread = threading.Thread(target=listen_loop)
        self.monitor_thread.start()

    def stop_listening(self):
        """Safely stop the background listening thread."""
        self.stop_signal.set()
        if self.monitor_thread:
            self.monitor_thread.join(timeout=1.0)

    def send_command(self, command: str):
        """Send a text command to the connected device."""
        if self.connection:
            self.connection.write((command + '\\r\\n').encode())

    def close(self):
        """Close the serial port."""
        self.stop_listening()
        if self.connection:
            self.connection.close()

# Example: Parsing GPS NMEA sentences
def parse_gps_sentence(sentence: str):
    """A simple parser for common GPS data sentences."""
    if not sentence.startswith('$'):
        return None
    parts = sentence.split(',')

    if sentence.startswith('$GPGGA'):  # Basic location and fix data
        if len(parts) > 9:
            return {
                'type': 'location',
                'time': parts[1],
                'lat': parts[2] + parts[3],  # Latitude and direction (N/S)
                'lon': parts[4] + parts[5],  # Longitude and direction (E/W)
                'altitude': parts[9] + parts[10]
            }
    elif sentence.startswith('$GPRMC'):  # Recommended minimum data
        if len(parts) > 7:
            return {
                'type': 'movement',
                'speed_knots': parts[7],
                'course': parts[8]
            }
    return None

# Using the monitor
print("Serial GPS Monitor Example")
print("=" * 40)

monitor = SerialMonitor()
ports = monitor.find_ports()
print(f"Available ports: {ports}")

if ports:
    # In a real scenario, you would connect to the correct port, e.g., ports[0]
    # monitor.connect(ports[0], 9600)

    def handle_incoming_data(data_line):
        parsed = parse_gps_sentence(data_line)
        if parsed:
            print(f"GPS Data: {parsed}")

    # monitor.start_listening(handle_incoming_data)
    # time.sleep(30)  # Listen for 30 seconds
    # monitor.stop_listening()
    # monitor.close()

    print("(Serial simulation skipped for example.)")

import RPi.GPIO as GPIO
import time

class ServoController:
    def __init__(self, control_pin: int):
        self.pin = control_pin
        GPIO.setmode(GPIO.BCM)
        GPIO.setup(self.pin, GPIO.OUT)

        # Standard servo PWM frequency is 50Hz (20ms period)
        self.pwm = GPIO.PWM(self.pin, 50)
        self.pwm.start(0)  # Start with 0 duty cycle
        self._current_angle = 90  # Assume starting at center

    def set_angle(self, angle: float):
        """Move the servo to a specific angle (typically 0 to 180 degrees)."""
        if angle < 0 or angle > 180:
            raise ValueError("Angle must be between 0 and 180 degrees.")

        # The magic formula: Duty Cycle = 2.5 + (angle / 18)
        # This gives a pulse width between 0.5ms (0°) and 2.5ms (180°)
        duty_cycle = 2.5 + (angle / 18.0)
        self.pwm.ChangeDutyCycle(duty_cycle)
        self._current_angle = angle
        time.sleep(0.5)  # Give the servo time to move
        self.pwm.ChangeDutyCycle(0)  # Stop sending pulse to avoid jitter

    def sweep(self, start_angle=0, end_angle=180, step=1, delay=0.02):
        """Smoothly sweep the servo between two angles."""
        for angle in range(start_angle, end_angle + step, step):
            self.set_angle(angle)
            time.sleep(delay)

    def cleanup(self):
        """Stop PWM and cleanup GPIO."""
        self.pwm.stop()
        GPIO.cleanup()

# Example usage
print("Servo Motor Control")
print("=" * 40)
try:
    servo = ServoController(control_pin=12)  # Assuming servo signal wire on GPIO12

    print("Moving to center (90°)")
    servo.set_angle(90)
    time.sleep(1)

    print("Moving to left (0°)")
    servo.set_angle(0)
    time.sleep(1)

    print("Sweeping from left to right...")
    servo.sweep(0, 180, 5, 0.05)

finally:
    servo.cleanup()

import usb.core
import usb.util

class USBDeviceFinder:
    """Finds and interacts with a USB device based on its Vendor ID and Product ID."""

    def __init__(self, vendor_id: int, product_id: int):
        self.vid = vendor_id
        self.pid = product_id
        self.device = None

    def find(self):
        """Locate the USB device."""
        self.device = usb.core.find(idVendor=self.vid, idProduct=self.pid)
        if self.device is None:
            raise ValueError("Device not found. Check connections.")
        print(f"Found device: {self.device}")
        return self.device

    def detach_kernel_driver(self, interface=0):
        """Sometimes the OS claims the device. This tells the OS to let go."""
        if self.device.is_kernel_driver_active(interface):
            try:
                self.device.detach_kernel_driver(interface)
                print("Detached kernel driver.")
            except usb.core.USBError as e:
                raise RuntimeError(f"Could not detach kernel driver: {e}")

    def write_data(self, endpoint: int, data: bytes):
        """Send data to a specific endpoint on the device."""
        # Endpoint 0x01 is typically a "bulk out" endpoint for sending data.
        bytes_written = self.device.write(endpoint, data)
        print(f"Sent {bytes_written} bytes.")
        return bytes_written

    def read_data(self, endpoint: int, size: int):
        """Read data from a specific endpoint."""
        # Endpoint 0x81 is typically a "bulk in" endpoint for receiving data.
        data = self.device.read(endpoint, size)
        print(f"Received {len(data)} bytes.")
        return data.tobytes()

# Note: This requires running with appropriate permissions (often as root/sudo).
print("USB Device Interaction Concept")
print("=" * 40)
print("This code shows the structure for talking to a custom USB device.")
print("Actual Vendor and Product IDs are needed to run it.")
# Example: finder = USBDeviceFinder(0x1234, 0xabcd)
# finder.find()

import paho.mqtt.client as mqtt
import time
import json

class IoT_Reporter:
    def __init__(self, broker_address="test.mosquitto.org", client_id=""):
        self.broker = broker_address
        self.client = mqtt.Client(client_id if client_id else mqtt.base62())
        self.client.on_connect = self._on_connect_callback
        self.client.on_message = self._on_message_callback

    def _on_connect_callback(self, client, userdata, flags, rc):
        """This runs when the connection to the broker is established."""
        if rc == 0:
            print("Connected successfully to MQTT broker!")
        else:
            print(f"Failed to connect, return code {rc}")

    def _on_message_callback(self, client, userdata, message):
        """This runs when a message is received on a subscribed topic."""
        print(f"Message on [{message.topic}]: {message.payload.decode()}")

    def connect(self):
        """Start the connection to the broker."""
        self.client.connect(self.broker, 1883, 60)
        self.client.loop_start()  # Start a background network thread

    def publish_sensor_data(self, topic: str, sensor_name: str, value: float, unit: str):
        """Format and send sensor data as a JSON message."""
        payload = {
            "sensor": sensor_name,
            "value": value,
            "unit": unit,
            "timestamp": time.time()
        }
        result = self.client.publish(topic, json.dumps(payload))
        if result.rc == mqtt.MQTT_ERR_SUCCESS:
            print(f"Published to {topic}")
        else:
            print(f"Publish failed: {result}")

    def subscribe(self, topic: str):
        """Listen for messages on a specific topic."""
        self.client.subscribe(topic)
        print(f"Subscribed to {topic}")

    def disconnect(self):
        """Cleanly disconnect from the broker."""
        self.client.loop_stop()
        self.client.disconnect()

# Example usage
print("MQTT IoT Reporting Example")
print("=" * 40)

reporter = IoT_Reporter("test.mosquitto.org")
reporter.connect()
time.sleep(1)  # Wait for connection

# Simulate publishing temperature every 5 seconds
try:
    for i in range(3):
        simulated_temp = 20.5 + i
        reporter.publish_sensor_data("home/livingroom/temperature", "bmp280", simulated_temp, "C")
        time.sleep(5)
finally:
    reporter.disconnect()

import cv2
import numpy as np
from ServoController import ServoController  # Using the class from earlier

class ColorTracker:
    def __init__(self, servo_pan_pin: int, servo_tilt_pin: int):
        # Initialize two servos: one for left-right (pan), one for up-down (tilt)
        self.servo_pan = ServoController(servo_pan_pin)
        self.servo_tilt = ServoController(servo_tilt_pin)
        self.cap = cv2.VideoCapture(0)

        # Define range of blue color in HSV (Hue, Saturation, Value)
        self.lower_blue = np.array([100, 150, 50])
        self.upper_blue = np.array([140, 255, 255])

        # Center position for the servos
        self.pan_angle = 90
        self.tilt_angle = 90
        self.servo_pan.set_angle(self.pan_angle)
        self.servo_tilt.set_angle(self.tilt_angle)

    def process_frame(self, frame):
        """Find the largest blue object in the frame and return its center."""
        # Convert from BGR to HSV color space
        hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)

        # Create a mask: white for blue pixels, black for everything else
        mask = cv2.inRange(hsv, self.lower_blue, self.upper_blue)

        # Find contours (outlines) of the white blobs in the mask
        contours, _ = cv2.findContours(mask, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)

        if contours:
            # Get the largest contour
            largest_contour = max(contours, key=cv2.contourArea)
            # Calculate the center of this contour
            M = cv2.moments(largest_contour)
            if M["m00"] != 0:
                center_x = int(M["m10"] / M["m00"])
                center_y = int(M["m01"] / M["m00"])
                return (center_x, center_y), largest_contour
        return None, None

    def update_servos(self, center_x, center_y, frame_width, frame_height):
        """Adjust servo angles based on object position."""
        # Convert pixel position to angle (0-180 range)
        new_pan = 180 - (center_x / frame_width) * 180  # Invert so object left moves servo left
        new_tilt = (center_y / frame_height) * 180

        # Smooth the movement: only move if the change is significant
        if abs(new_pan - self.pan_angle) > 2:
            self.pan_angle = max(0, min(180, new_pan))
            self.servo_pan.set_angle(self.pan_angle)
        if abs(new_tilt - self.tilt_angle) > 2:
            self.tilt_angle = max(0, min(180, new_tilt))
            self.servo_tilt.set_angle(self.tilt_angle)

    def run(self):
        """Main loop to capture video and track."""
        print("Starting color tracker. Press 'q' to quit.")
        while True:
            ret, frame = self.cap.read()
            if not ret:
                break

            frame = cv2.flip(frame, 1)  # Mirror the image
            height, width, _ = frame.shape

            center, contour = self.process_frame(frame)

            if center:
                # Draw a circle and rectangle on the object
                cv2.circle(frame, center, 10, (0, 255, 0), -1)
                x, y, w, h = cv2.boundingRect(contour)
                cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 255, 0), 2)

                # Move the servos to follow it
                self.update_servos(center[0], center[1], width, height)

            # Show the video feed
            cv2.imshow('Color Tracker', frame)

            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

        self.cap.release()
        cv2.destroyAllWindows()
        self.servo_pan.cleanup()
        self.servo_tilt.cleanup()

# To run this:
# tracker = ColorTracker(servo_pan_pin=12, servo_tilt_pin=13)
# tracker.run()
print("Computer Vision Servo Tracker")
print("=" * 40)
print("This code combines OpenCV for seeing a blue object with servos to follow it.")