組み込みシステム・IoT完全ガイド: MCUからエッジAIまで

// STM32 HALライブラリを使ったGPIO制御
#include "stm32f4xx_hal.h"

// LED トグル（ピン PA5）
HAL_GPIO_TogglePin(GPIOA, GPIO_PIN_5);
HAL_Delay(500);

// ボタン読み取り（ピン PC13、アクティブLOW）
if (HAL_GPIO_ReadPin(GPIOC, GPIO_PIN_13) == GPIO_PIN_RESET) {
    // ボタン押下時の処理
    HAL_GPIO_WritePin(GPIOA, GPIO_PIN_5, GPIO_PIN_SET);
}

// レジスタ直接操作（STM32F4）
// GPIOAクロック有効化
RCC->AHB1ENR |= RCC_AHB1ENR_GPIOAEN;

// PA5を出力に設定（MODERレジスタ）
GPIOA->MODER &= ~(0x3 << (5 * 2));  // ビットクリア
GPIOA->MODER |=  (0x1 << (5 * 2));  // 出力モード設定

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

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

// 誤り: コンパイラが最適化で除去する可能性あり
uint32_t *reg = (uint32_t *)0x40020000;
*reg = 0x01;

// 正解: volatileを使用
volatile uint32_t *reg = (volatile uint32_t *)0x40020000;
*reg = 0x01;

// ISRで変更されるグローバル変数もvolatileが必要
volatile uint8_t button_pressed = 0;

// EXTI割り込みコールバック（HAL方式）
void HAL_GPIO_EXTI_Callback(uint16_t GPIO_Pin) {
    if (GPIO_Pin == GPIO_PIN_13) {
        button_pressed = 1;
        // ISR内では最小限の処理のみ
        // HAL_Delay()などのブロッキング関数は使用禁止
    }
}

// タイマー割り込みコールバック（1msごとに呼び出し）
void HAL_TIM_PeriodElapsedCallback(TIM_HandleTypeDef *htim) {
    if (htim->Instance == TIM2) {
        system_tick++;
    }
}

// PWM開始（TIM3チャンネル1）
HAL_TIM_PWM_Start(&htim3, TIM_CHANNEL_1);

// デューティ比50%設定（ARR=999のとき CCR=500）
__HAL_TIM_SET_COMPARE(&htim3, TIM_CHANNEL_1, 500);

// サーボモータ角度制御（0〜180度）
void servo_set_angle(uint8_t angle) {
    // 1ms〜2msのパルス幅をタイマーカウンタ値に変換
    uint32_t pulse = 500 + (angle * 1000 / 180);
    __HAL_TIM_SET_COMPARE(&htim3, TIM_CHANNEL_1, pulse);
}

// UART文字列送信
char msg[] = "Hello, UART!\r\n";
HAL_UART_Transmit(&huart2, (uint8_t*)msg, strlen(msg), 100);

// UART受信（割り込み方式）
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送受信（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

// I2CでMPU6050加速度センサー読み取り
#define MPU6050_ADDR  0xD0  // 7ビットアドレス0x68、書き込み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];

// 実際の加速度（g単位）= raw / 16384.0（±2g設定時）
float ax = accel_x / 16384.0f;

// CANメッセージ送信（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"

// LEDブリンクタスク
void vTaskLED(void *pvParameters) {
    while(1) {
        HAL_GPIO_TogglePin(LED_GPIO_Port, LED_Pin);
        vTaskDelay(pdMS_TO_TICKS(500));  // 500ms待機（CPU譲渡）
    }
}

// センサー読み取りタスク
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));

    // タスク生成（関数名, タスク名, スタックサイズ, パラメータ, 優先度, ハンドル）
    xTaskCreate(vTaskLED,    "LED",    128, NULL,        1, NULL);
    xTaskCreate(vTaskSensor, "Sensor", 256, xTempQueue,  2, NULL);

    vTaskStartScheduler();  // RTOSスケジューラ起動（ここから戻らない）
    while(1);
}

SemaphoreHandle_t xMutex;
SemaphoreHandle_t xSemaphore;

// ミューテックスで共有リソースを保護
xMutex = xSemaphoreCreateMutex();

void vTaskA(void *pvParameters) {
    while(1) {
        if (xSemaphoreTake(xMutex, portMAX_DELAY) == pdTRUE) {
            // クリティカルセクション
            shared_data++;
            xSemaphoreGive(xMutex);
        }
    }
}

// バイナリセマフォでISR-タスク同期
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 -- 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接続:', 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ステーション初期化
#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)

# 50Hz PWMでサーボモータ制御
pwm = GPIO.PWM(18, 50)
pwm.start(7.5)  # ニュートラル位置（90度）

def set_servo_angle(angle):
    """サーボの角度を0〜180度で設定"""
    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}度")
        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超音波距離センサー
sensor = DistanceSensor(echo=24, trigger=23)
while True:
    dist = sensor.distance * 100  # cm単位
    print(f"距離: {dist:.1f} cm")
    if dist < 20:
        led.blink(on_time=0.1, off_time=0.1)
    time.sleep(0.5)

# Raspberry Pi向けPython IoTアプリ Dockerfile
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 -- フルIoTスタック
version: '3'
services:
  sensor-app:
    build: .
    privileged: true # GPIO接続に必要
    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画像分類推論（Raspberry Pi）
import tflite_runtime.interpreter as tflite
import numpy as np
from PIL import Image
import time

# インタプリタのロードと初期化
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_details[0]['shape'])   # [1, 224, 224, 3]
print("入力型:", input_details[0]['dtype'])     # uint8（量子化モデル）

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"ラベル: {labels[top_idx]}, スコア: {output[0][top_idx]}")
    print(f"推論時間: {elapsed:.1f}ms")

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

classify_image('test.jpg', labels)

# 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_name}, 形状: {input_shape}")

# 画像前処理
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)   # バッチ次元追加

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

# Python Paho-MQTTクライアント
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"接続コード: {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)

# 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アップデート
#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,  // サーバー証明書検証
        },
    };

    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暗号化（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ビット鍵

    uint8_t iv_copy[16];
    memcpy(iv_copy, iv, 16);  // IVは暗号化中に変化

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