Split View: 임베디드 시스템 & IoT 완전 가이드: MCU부터 엣지 AI까지

// GPIO control using STM32 HAL library
#include "stm32f4xx_hal.h"

// Toggle LED on pin PA5
HAL_GPIO_TogglePin(GPIOA, GPIO_PIN_5);
HAL_Delay(500);

// Read button on pin PC13 (active LOW)
if (HAL_GPIO_ReadPin(GPIOC, GPIO_PIN_13) == GPIO_PIN_RESET) {
    // Button is pressed — turn LED on
    HAL_GPIO_WritePin(GPIOA, GPIO_PIN_5, GPIO_PIN_SET);
}

// Direct register access (STM32F4)
// Enable GPIOA peripheral clock
RCC->AHB1ENR |= RCC_AHB1ENR_GPIOAEN;

// Set PA5 as output (MODER register)
GPIOA->MODER &= ~(0x3 << (5 * 2));  // Clear bits
GPIOA->MODER |=  (0x1 << (5 * 2)); // Output mode

// Set PA5 HIGH
GPIOA->BSRR = (1 << 5);

// Set PA5 LOW
GPIOA->BSRR = (1 << (5 + 16));

// WRONG: compiler may optimize this out entirely
uint32_t *reg = (uint32_t *)0x40020000;
*reg = 0x01;

// CORRECT: declare as volatile
volatile uint32_t *reg = (volatile uint32_t *)0x40020000;
*reg = 0x01;

// Global variables modified in ISRs must also be volatile
volatile uint8_t button_pressed = 0;

// External interrupt callback (HAL style)
void HAL_GPIO_EXTI_Callback(uint16_t GPIO_Pin) {
    if (GPIO_Pin == GPIO_PIN_13) {
        button_pressed = 1;
        // Keep ISRs short — never call blocking functions like HAL_Delay()
    }
}

// Timer period elapsed callback (called every 1ms)
void HAL_TIM_PeriodElapsedCallback(TIM_HandleTypeDef *htim) {
    if (htim->Instance == TIM2) {
        system_tick++;
    }
}

// Start PWM on TIM3 Channel 1
HAL_TIM_PWM_Start(&htim3, TIM_CHANNEL_1);

// Set 50% duty cycle (when ARR = 999, set CCR = 500)
__HAL_TIM_SET_COMPARE(&htim3, TIM_CHANNEL_1, 500);

// Servo motor control function (0 to 180 degrees)
void servo_set_angle(uint8_t angle) {
    // Map angle to timer compare value for 1ms~2ms pulse width
    uint32_t pulse = 500 + (angle * 1000 / 180);
    __HAL_TIM_SET_COMPARE(&htim3, TIM_CHANNEL_1, pulse);
}

// Transmit a string over UART
char msg[] = "Hello, UART!\r\n";
HAL_UART_Transmit(&huart2, (uint8_t*)msg, strlen(msg), 100);

// Receive a byte using interrupt mode
uint8_t rx_byte;
HAL_UART_Receive_IT(&huart2, &rx_byte, 1);

void HAL_UART_RxCpltCallback(UART_HandleTypeDef *huart) {
    if (huart->Instance == USART2) {
        rx_buffer[rx_index++] = rx_byte;
        HAL_UART_Receive_IT(&huart2, &rx_byte, 1);
    }
}

// SPI transmit/receive (reading Flash ID)
uint8_t tx_data[] = {0x9F, 0x00};
uint8_t rx_data[2];

HAL_GPIO_WritePin(CS_GPIO_Port, CS_Pin, GPIO_PIN_RESET);  // CS LOW
HAL_SPI_TransmitReceive(&hspi1, tx_data, rx_data, 2, 100);
HAL_GPIO_WritePin(CS_GPIO_Port, CS_Pin, GPIO_PIN_SET);    // CS HIGH

// Read MPU6050 accelerometer over I2C
#define MPU6050_ADDR  0xD0  // 7-bit address 0x68, write = 0xD0
#define ACCEL_XOUT_H  0x3B

uint8_t data[6];
HAL_I2C_Mem_Read(&hi2c1, MPU6050_ADDR, ACCEL_XOUT_H,
                  I2C_MEMADD_SIZE_8BIT, data, 6, 100);

int16_t accel_x = (data[0] << 8) | data[1];
int16_t accel_y = (data[2] << 8) | data[3];
int16_t accel_z = (data[4] << 8) | data[5];

// Convert to real acceleration in g (at ±2g setting)
float ax = accel_x / 16384.0f;

// Send a CAN message (STM32 HAL)
CAN_TxHeaderTypeDef tx_header;
uint8_t tx_data[8] = {0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08};
uint32_t tx_mailbox;

tx_header.StdId = 0x123;
tx_header.IDE   = CAN_ID_STD;
tx_header.RTR   = CAN_RTR_DATA;
tx_header.DLC   = 8;

HAL_CAN_AddTxMessage(&hcan1, &tx_header, tx_data, &tx_mailbox);

#include "FreeRTOS.h"
#include "task.h"
#include "semphr.h"
#include "queue.h"

// Task function — blinks an LED
void vTaskLED(void *pvParameters) {
    while(1) {
        HAL_GPIO_TogglePin(LED_GPIO_Port, LED_Pin);
        vTaskDelay(pdMS_TO_TICKS(500));  // Yield CPU for 500ms
    }
}

// Task function — reads a sensor and posts to a queue
void vTaskSensor(void *pvParameters) {
    QueueHandle_t xQueue = (QueueHandle_t)pvParameters;
    float temperature;

    while(1) {
        temperature = read_temperature_sensor();
        xQueueSend(xQueue, &temperature, portMAX_DELAY);
        vTaskDelay(pdMS_TO_TICKS(1000));
    }
}

int main(void) {
    HAL_Init();
    SystemClock_Config();

    QueueHandle_t xTempQueue = xQueueCreate(10, sizeof(float));

    // Create tasks (function, name, stack depth, param, priority, handle)
    xTaskCreate(vTaskLED,    "LED",    128, NULL,        1, NULL);
    xTaskCreate(vTaskSensor, "Sensor", 256, xTempQueue,  2, NULL);

    vTaskStartScheduler();  // Hand control to the RTOS — never returns
    while(1);
}

SemaphoreHandle_t xMutex;
SemaphoreHandle_t xSemaphore;

// Mutex — protects a shared resource
xMutex = xSemaphoreCreateMutex();

void vTaskA(void *pvParameters) {
    while(1) {
        if (xSemaphoreTake(xMutex, portMAX_DELAY) == pdTRUE) {
            // Critical section
            shared_data++;
            xSemaphoreGive(xMutex);
        }
    }
}

// Binary semaphore — synchronizes ISR to task
xSemaphore = xSemaphoreCreateBinary();

void HAL_GPIO_EXTI_Callback(uint16_t GPIO_Pin) {
    BaseType_t xHigherPriorityTaskWoken = pdFALSE;
    xSemaphoreGiveFromISR(xSemaphore, &xHigherPriorityTaskWoken);
    portYIELD_FROM_ISR(xHigherPriorityTaskWoken);
}

#include <DHT.h>
#include <LiquidCrystal_I2C.h>

#define DHTPIN  2
#define DHTTYPE DHT22

DHT dht(DHTPIN, DHTTYPE);
LiquidCrystal_I2C lcd(0x27, 16, 2);

void setup() {
    Serial.begin(9600);
    dht.begin();
    lcd.init();
    lcd.backlight();
    lcd.print("IoT Weather Box");
    delay(2000);
    lcd.clear();
}

void loop() {
    float temp = dht.readTemperature();
    float hum  = dht.readHumidity();

    if (isnan(temp) || isnan(hum)) {
        Serial.println("DHT read error!");
        return;
    }

    lcd.setCursor(0, 0);
    lcd.print("Temp: ");
    lcd.print(temp, 1);
    lcd.print(" C  ");

    lcd.setCursor(0, 1);
    lcd.print("Hum:  ");
    lcd.print(hum, 1);
    lcd.print(" %  ");

    Serial.print("T="); Serial.print(temp);
    Serial.print(" H="); Serial.println(hum);

    delay(2000);
}

# MicroPython ESP32 — publish sensor data over MQTT
import time
import network
from umqtt.simple import MQTTClient
from machine import Pin, ADC

def connect_wifi(ssid, password):
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        wlan.connect(ssid, password)
        while not wlan.isconnected():
            time.sleep(0.5)
    print('WiFi connected:', wlan.ifconfig())

def on_message(topic, msg):
    print('Topic:', topic, 'Message:', msg)

connect_wifi('MySSID', 'MyPassword')

adc = ADC(Pin(34))
adc.atten(ADC.ATTN_11DB)

client = MQTTClient('esp32_sensor', 'broker.hivemq.com', port=1883)
client.set_callback(on_message)
client.connect()
client.subscribe(b'device/control')

while True:
    raw = adc.read()
    voltage = raw * 3.3 / 4095
    temp_approx = (voltage - 0.5) * 100
    payload = '{:.1f}'.format(temp_approx)
    client.publish(b'sensor/temperature', payload.encode())
    client.check_msg()
    time.sleep(5)

// ESP-IDF WiFi station initialization
#include "esp_wifi.h"
#include "esp_event.h"
#include "nvs_flash.h"

#define WIFI_SSID "MyNetwork"
#define WIFI_PASS "MyPassword"

static void wifi_event_handler(void *arg, esp_event_base_t event_base,
                                int32_t event_id, void *event_data) {
    if (event_base == WIFI_EVENT && event_id == WIFI_EVENT_STA_START) {
        esp_wifi_connect();
    } else if (event_base == IP_EVENT && event_id == IP_EVENT_STA_GOT_IP) {
        ip_event_got_ip_t *event = (ip_event_got_ip_t *)event_data;
        ESP_LOGI("WiFi", "Got IP: " IPSTR, IP2STR(&event->ip_info.ip));
    }
}

void wifi_init_sta(void) {
    nvs_flash_init();
    esp_netif_init();
    esp_event_loop_create_default();
    esp_netif_create_default_wifi_sta();

    wifi_init_config_t cfg = WIFI_INIT_CONFIG_DEFAULT();
    esp_wifi_init(&cfg);

    esp_event_handler_register(WIFI_EVENT, ESP_EVENT_ANY_ID, wifi_event_handler, NULL);
    esp_event_handler_register(IP_EVENT, IP_EVENT_STA_GOT_IP, wifi_event_handler, NULL);

    wifi_config_t wifi_config = {
        .sta = { .ssid = WIFI_SSID, .password = WIFI_PASS },
    };
    esp_wifi_set_mode(WIFI_MODE_STA);
    esp_wifi_set_config(WIFI_IF_STA, &wifi_config);
    esp_wifi_start();
}

import RPi.GPIO as GPIO
import time

GPIO.setmode(GPIO.BCM)
GPIO.setup(18, GPIO.OUT)

pwm = GPIO.PWM(18, 50)  # 50Hz for servo
pwm.start(7.5)          # Neutral position (90 degrees)

def set_servo_angle(angle):
    """Set servo angle from 0 to 180 degrees."""
    duty = 2.5 + (angle / 180.0) * 10.0
    pwm.ChangeDutyCycle(duty)
    time.sleep(0.3)

try:
    for angle in [0, 45, 90, 135, 180]:
        set_servo_angle(angle)
        print(f"Angle: {angle} degrees")
        time.sleep(1)
finally:
    pwm.stop()
    GPIO.cleanup()

from gpiozero import LED, Button, DistanceSensor
import time

led    = LED(17)
button = Button(4)

button.when_pressed  = led.on
button.when_released = led.off

# HC-SR04 ultrasonic distance sensor
sensor = DistanceSensor(echo=24, trigger=23)
while True:
    dist = sensor.distance * 100  # convert to cm
    print(f"Distance: {dist:.1f} cm")
    if dist < 20:
        led.blink(on_time=0.1, off_time=0.1)
    time.sleep(0.5)

# Dockerfile for a Raspberry Pi IoT app
FROM python:3.11-slim-bullseye

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
CMD ["python", "sensor_server.py"]

# docker-compose.yml — full IoT stack
version: '3'
services:
  sensor-app:
    build: .
    privileged: true # Required for GPIO access
    volumes:
      - /dev:/dev
    restart: unless-stopped

  influxdb:
    image: influxdb:2.7
    ports:
      - '8086:8086'
    volumes:
      - influx_data:/var/lib/influxdb2

  grafana:
    image: grafana/grafana:latest
    ports:
      - '3000:3000'
    volumes:
      - grafana_data:/var/lib/grafana

volumes:
  influx_data:
  grafana_data:

# TFLite image classification on Raspberry Pi
import tflite_runtime.interpreter as tflite
import numpy as np
from PIL import Image
import time

# Load and initialize the interpreter
interpreter = tflite.Interpreter(
    model_path='mobilenet_v2_1.0_224_quant.tflite',
    num_threads=4
)
interpreter.allocate_tensors()

input_details  = interpreter.get_input_details()
output_details = interpreter.get_output_details()

print("Input shape:", input_details[0]['shape'])   # [1, 224, 224, 3]
print("Input dtype:", input_details[0]['dtype'])   # uint8 for quantized model

def classify_image(image_path, labels):
    img = Image.open(image_path).convert('RGB').resize((224, 224))
    input_data = np.expand_dims(np.array(img), axis=0)

    interpreter.set_tensor(input_details[0]['index'], input_data)

    start = time.time()
    interpreter.invoke()
    elapsed = (time.time() - start) * 1000

    output  = interpreter.get_tensor(output_details[0]['index'])
    top_idx = np.argmax(output[0])

    print(f"Label: {labels[top_idx]}, Score: {output[0][top_idx]}")
    print(f"Inference time: {elapsed:.1f}ms")

with open('labels.txt') as f:
    labels = [line.strip() for line in f.readlines()]

classify_image('test.jpg', labels)

# Running a custom model with ONNX Runtime
import onnxruntime as ort
import numpy as np
from PIL import Image

session = ort.InferenceSession(
    'model.onnx',
    providers=['CPUExecutionProvider']
)

input_name  = session.get_inputs()[0].name
output_name = session.get_outputs()[0].name
input_shape = session.get_inputs()[0].shape
print(f"Input: {input_name}, Shape: {input_shape}")

# Preprocess image
img       = Image.open('test.jpg').resize((224, 224))
img_array = np.array(img).astype(np.float32) / 255.0
img_array = np.transpose(img_array, (2, 0, 1))  # HWC to CHW
img_array = np.expand_dims(img_array, axis=0)   # Add batch dimension

result = session.run([output_name], {input_name: img_array})
print("Output shape:", result[0].shape)

# Python Paho-MQTT client
import paho.mqtt.client as mqtt
import json
import time

BROKER    = "broker.hivemq.com"
PORT      = 1883
TOPIC_PUB = "home/sensor/living_room"
TOPIC_SUB = "home/control/#"

def on_connect(client, userdata, flags, rc):
    print(f"Connected with code: {rc}")
    client.subscribe(TOPIC_SUB)

def on_message(client, userdata, msg):
    payload = json.loads(msg.payload.decode())
    print(f"Topic: {msg.topic}, Payload: {payload}")

client = mqtt.Client("python_sensor_01")
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

while True:
    data = {
        "temperature": 23.5,
        "humidity": 60.2,
        "timestamp": int(time.time())
    }
    client.publish(TOPIC_PUB, json.dumps(data))
    time.sleep(10)

# Write sensor data to InfluxDB 2.x
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

INFLUX_URL    = "http://localhost:8086"
INFLUX_TOKEN  = "your_token_here"
INFLUX_ORG    = "myorg"
INFLUX_BUCKET = "iot_sensors"

client    = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)
write_api = client.write_api(write_options=SYNCHRONOUS)

point = (
    Point("environment")
    .tag("location", "living_room")
    .tag("device", "esp32_01")
    .field("temperature", 23.5)
    .field("humidity", 60.2)
    .field("co2_ppm", 650)
)
write_api.write(bucket=INFLUX_BUCKET, record=point)

// ESP-IDF HTTPS OTA update
#include "esp_ota_ops.h"
#include "esp_https_ota.h"

void ota_task(void *pvParameter) {
    esp_https_ota_config_t ota_config = {
        .http_config = &(esp_http_client_config_t){
            .url      = "https://myserver.com/firmware.bin",
            .cert_pem = server_cert_pem,  // Server certificate verification
        },
    };

    esp_err_t ret = esp_https_ota(&ota_config);
    if (ret == ESP_OK) {
        esp_restart();
    } else {
        ESP_LOGE("OTA", "Update failed: %s", esp_err_to_name(ret));
    }
    vTaskDelete(NULL);
}

// mbedTLS AES-128 CBC encryption (STM32/ESP32)
#include "mbedtls/aes.h"

void aes_encrypt_data(uint8_t *plaintext, uint8_t *ciphertext,
                       uint8_t *key, uint8_t *iv, size_t len) {
    mbedtls_aes_context ctx;
    mbedtls_aes_init(&ctx);
    mbedtls_aes_setkey_enc(&ctx, key, 128);  // 128-bit key

    uint8_t iv_copy[16];
    memcpy(iv_copy, iv, 16);  // IV is mutated during encryption

    mbedtls_aes_crypt_cbc(&ctx, MBEDTLS_AES_ENCRYPT, len,
                           iv_copy, plaintext, ciphertext);
    mbedtls_aes_free(&ctx);
}