---
# System prepended metadata

title: 智慧杯墊 Intelligent Coaster

---

# 智慧杯墊 Intelligent Coaster

## 項目概述 (Project Overview)
隨著智慧健康裝置的普及，人們逐漸重視如何利用科技改善日常生活。本專案目的在開發一款多功能的智慧杯墊，以提升用戶的飲水習慣與生活品質。該杯墊基於樹莓派(Raspberry pi)平台，整合物聯網技術，具備飲水量偵測、震動攪拌、測量合適溫度、飲水紀錄、智慧提醒及健康建議等功能，並利用 LINE Bot 進行提醒與記錄管理，適應多種使用場景，促進健康管理。

## 系統特色與創新亮點
智慧杯墊結合 Raspberry Pi + 感測器 + LINE Bot，具備以下特色：
### 多模態偵測：
整合重量感測器與紅外線溫度感測器，自動判斷喝水量與飲品溫度。
### 即時互動：
透過 LED 燈光與震動馬達，提供直覺式回饋（如溫度顏色提示、攪拌功能）。
### 個人化建議：
依據使用者體重、運動習慣與天氣資訊，自動計算每日建議飲水量。
### 雲端整合：
使用 LINE Bot 作為主要互動介面，記錄與提醒同步進行，降低學習成本。

## 成果展示（Demonstration)
[智慧杯墊demo影片](https://youtube.com/shorts/MG-9ypqVU5U?si=XqYuKSWZq3CoWHP6)
### 功能1 : 飲水量與溫度偵測
每次杯墊偵測到使用者拿起水杯並放下後，即會自動計算喝水量及偵測溫度。
![S__131457029_0](https://hackmd.io/_uploads/SkyesHl2le.jpg)
### 功能2 : 偵測溫度並以光學展示
使用者在LINE Bot輸入"current temperature"後，杯墊會自動偵測飲品溫度，並依照溫度適不適合飲用發出不同顏色的光。
![S__131457028_0](https://hackmd.io/_uploads/S1qR7Ux2le.jpg)
![image](https://hackmd.io/_uploads/r1YIZG72lx.png)

### 功能3 : 智慧攪拌
使用者在LINE Bot輸入"stir"後，會被詢問要攪拌幾秒，回答完成後杯墊會開始攪拌飲品。
![S__131457030_0](https://hackmd.io/_uploads/ryNNVLg2ee.jpg)
### 功能4 : 喝水提醒
使用者在LINE Bot輸入"remind time"後，會被詢問希望於何時被提醒，回答完成後杯墊將會在那個時間傳訊息與發光提醒使用者喝水。
![S__131457031_0](https://hackmd.io/_uploads/rJ10E8xnge.jpg)
### 功能5 : 每日飲水量計算
使用者在LINE Bot輸入"water intake"後，會被詢問體重及一周運動次數，回答完成後系統會自動計算每日建議飲水量。
![S__131457033_0](https://hackmd.io/_uploads/rJ0SrUe2xe.jpg)
### 功能6 : 飲水建議
使用者在LINE Bot輸入"recommend"後，系統將根據氣溫、濕度和當前時間給予使用者飲水建議。
![S__131457032_0](https://hackmd.io/_uploads/Bk1eULl2le.jpg)
### 功能7 : 紀錄查詢
使用者在LINE Bot輸入"recommend"後，將收到近期喝水量及溫度的紀錄。
![S__131457034_0](https://hackmd.io/_uploads/H1NHUIx3ee.jpg)


## 必備材料與工具 (Required Materials and Tools)

### 硬體
* 樹莓派 Raspberry Pi Zero WH 開發板
* SD Card 32GB
* MLX90614 非接觸式 紅外線測溫感測器
* 外接電池座+電池*4
* 1kg 負載感測器
* HX711 秤重感測器模組
* 5V RGB WS2812/WS2813 防水型矽膠軟式燈條
* 震動馬達模組
* 協助組合之雙面膠/膠帶
* 迷你麵包板
* 杜邦線(母母&公母&公公)
* 紙板/任何好組裝之杯墊材料
(原先使用矽膠片，因無法妥善沾黏組合換為紙板。
仍建議使用防水、防滑之材質。)

### 軟體
* Python 3
* Flask
* SQLite 資料庫
* LINE Messaging API

## 硬體與電路設計 (Hardware and Circuit Design)
### 電路圖
![S__7602181](https://hackmd.io/_uploads/SJX8dN3L1x.jpg)

### 照片與示意圖
![S__105365507_0](https://hackmd.io/_uploads/HkyOA4nIkg.jpg)
![S__105365509_0](https://hackmd.io/_uploads/ByzdA4nUye.jpg)


## 步驟教學 (Step-by-Step Tutorial)
1. 先將硬體照著電路圖組裝完畢。
2. 製作杯墊外殼，並將重量感測器放在中間。
將溫度感測器和震動馬達黏在上蓋，溫度感測器上方開洞。
3. 依照[https://www.raspberrypi.com/documentation/computers/getting-started.html](https://) 燒錄完成SD卡，並可以在介面操控樹莓派。
4. 前往[https://ngrok.com/](https://) 申請帳號登入，並記錄自己的authtoken。
5. 依照 [https://download.ngrok.com/raspberry-pi?tab=download](https://) 在樹莓派上下載並執行ngrok，記錄自己的公開 URL。
6. 在[https://openweathermap.org/](https://) 上註冊並獲取自己的APIKEY。
7. 在[https://manager.line.biz/](https://) 上註冊並創建一個官方帳號。
8. 在[https://developers.line.biz/zh-hant/](https://) 上註冊並創建一個Messaging API的LINE Channel。
9. 在 LINE Business Manager 設定中連接 Messaging API Channel 和官方帳號，將兩者綁定。
10. 在 LINE developers 中，紀錄Channel Access Token，並將ngrok的公開 URL輸入。
11. 下載執行程式需要的庫，在樹莓派的終端機輸入:
```bash=
sudo apt update
sudo apt upgrade

sudo apt install python3-pip

sudo pip3 install adafruit-blinka
sudo pip3 install adafruit-circuitpython-neopixel
sudo pip3 install adafruit-circuitpython-mlx90614

sudo pip3 install hx711
sudo pip3 install Flask
sudo apt install python3-rpi.gpio
sudo pip3 install requests
```
12. 在樹莓派的python上輸入 :
```python=
# Main Imports
import time
import board
import busio
import neopixel
import sqlite3
import requests
import threading
import RPi.GPIO as GPIO
from adafruit_mlx90614 import MLX90614
from flask import Flask, request, jsonify
from datetime import datetime
from hx711 import HX711

app = Flask(__name__)

# Database Setup
db_path = 'water_intake.db'
conn = sqlite3.connect(db_path, check_same_thread=False)

# Hardware and Sensor Setup
# I2C Sensor (Temperature)
i2c = busio.I2C(board.SCL, board.SDA)
sensor = MLX90614(i2c)

# LED Strip
LED_PIN = board.D18
NUM_PIXELS = 10
ORDER = neopixel.GRB
pixels = neopixel.NeoPixel(LED_PIN, NUM_PIXELS, brightness=0.5, auto_write=False, pixel_order=ORDER)

# HX711 (Weight Sensor)
DT = 5  # Replace with your actual DT pin
SCK = 6  # Replace with your actual SCK pin
hx = HX711(DT, SCK)
hx.set_reading_format("MSB", "MSB")
CALIBRATION_FACTOR = YOUR_CALIBRATION_FACTOR
hx.tare(times=20)  # Tare the scale

# Motor Setup
MOTOR_PIN = 17
GPIO.setmode(GPIO.BCM)
GPIO.setup(MOTOR_PIN, GPIO.OUT)

# Global Variables
remind_time = None
last_weight = 0
is_drinking = False
weight_history = []
drinking_threshold = 10
stability_time = 3
check_interval = 0.5
MIN_DETECTABLE_WEIGHT = 10  # Minimum weight threshold in grams


# Utility Functions
def flash_led(color, flashes):
    """Flash the LED strip a specified number of times."""
    for _ in range(flashes):
        pixels.fill(color)
        pixels.show()
        time.sleep(0.5)
        pixels.fill((0, 0, 0))
        pixels.show()
        time.sleep(0.5)


def set_led_color(r, g, b, duration=3):
    """Set the LED strip to a specific color for a duration."""
    pixels.fill((r, g, b))
    pixels.show()
    time.sleep(duration)
    pixels.fill((0, 0, 0))
    pixels.show()


def get_temperature():
    """Get the temperature from the MLX90614 sensor."""
    return sensor.object_temperature


# Weight Monitoring
def get_stable_weight(hx, samples=20):
    """Get a stable weight reading from the HX711."""
    try:
        readings = [hx.get_weight_A(1) for _ in range(samples)]
        readings = [abs(r) for r in readings]
        readings.sort()
        trimmed_readings = readings[int(0.1 * samples):int(0.9 * samples)]
        return sum(trimmed_readings) / len(trimmed_readings)
    except Exception as e:
        print(f"Error reading weight: {e}")
        return None


def get_weight():
    """Retrieve the current weight in grams."""
    try:
        raw_weight = get_stable_weight(hx, samples=20)
        if raw_weight is not None:
            weight_in_grams = int((raw_weight / 100) / CALIBRATION_FACTOR)
            return max(weight_in_grams, MIN_DETECTABLE_WEIGHT)  # Ensure minimum detectable weight
        return 0
    except Exception as e:
        print(f"Error reading weight: {e}")
        return 0


def monitor_weight():
    """Monitor weight changes and detect drinking events."""
    global last_weight, is_drinking, weight_history
    current_weight = get_weight()
    weight_history.append(current_weight)

    if len(weight_history) > stability_time:
        weight_history.pop(0)

    average_weight = sum(weight_history) / len(weight_history)
    print(f"Weight monitoring: Last: {last_weight}, Current: {current_weight}, Avg: {average_weight:.2f}")

    if current_weight == 0 or (average_weight < drinking_threshold and not is_drinking):
        is_drinking = True
        print("Started drinking.")
    elif is_drinking and average_weight >= drinking_threshold:
        drink_amount = last_weight - average_weight
        if drink_amount > 0:
            print(f"Detected drink amount: {drink_amount} grams")
            store_data(drink_amount, get_temperature())
        is_drinking = False
        print("Finished drinking.")

    if not is_drinking:
        last_weight = average_weight

    time.sleep(check_interval)


# Reminder Functionality
def monitor_reminder():
    """Monitor and trigger reminders based on the set time."""
    global remind_time
    while True:
        if remind_time:
            current_time = datetime.now().strftime("%H:%M")
            if current_time == remind_time:
                flash_led((0, 0, 255), 5)
                remind_time = None
        time.sleep(1)


# Database Functions
def store_data(weight, temperature):
    """Store water intake data in the database."""
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS records
                 (id INTEGER PRIMARY KEY AUTOINCREMENT,
                  timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                  weight REAL,
                  temperature REAL)''')
    c.execute('INSERT INTO records (weight, temperature) VALUES (?, ?)', (weight, temperature))
    conn.commit()


def fetch_records(limit=10):
    """Fetch the most recent records from the database."""
    c = conn.cursor()
    c.execute('SELECT timestamp, weight, temperature FROM records ORDER BY timestamp DESC LIMIT ?', (limit,))
    records = c.fetchall()
    return records


# Motor Control
def start_motor(duration=2):
    """Start the motor for a specified duration."""
    pwm = GPIO.PWM(MOTOR_PIN, 300)
    pwm.start(100)
    time.sleep(duration)
    pwm.stop()


# Weather Data
def get_weather_data():
    """Fetch weather data from OpenWeatherMap API."""
    API_KEY = 'YOUR_API_KEY'
    CITY = 'Taipei'
    url = f"http://api.openweathermap.org/data/2.5/weather?q={CITY}&appid={API_KEY}&units=metric"
    response = requests.get(url)
    if response.status_code == 200:
        data = response.json()
        return data['main']['temp'], data['main']['humidity']
    else:
        print("Error fetching weather data.")
        return None, None


# Background Threads
threading.Thread(target=monitor_reminder, daemon=True).start()
threading.Thread(target=monitor_weight, daemon=True).start()

# LINE API configuration
LINE_CHANNEL_ACCESS_TOKEN = 'YOUR_LINE_CHANNEL_ACCESS_TOKEN'
LINE_USER_ID = 'YOUR_LINE_USER_ID'

# Helper function to send notifications via LINE
def notify_user(message):
    url = 'https://api.line.me/v2/bot/message/push'
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {LINE_CHANNEL_ACCESS_TOKEN}'
    }
    data = {
        'to': LINE_USER_ID,
        'messages': [{'type': 'text', 'text': message}]
    }
    response = requests.post(url, json=data, headers=headers)
    if response.status_code != 200:
        print(f"Failed to send notification: {response.status_code}, {response.text}")

# Helper function to send a reply via LINE
def send_line_reply(reply_token, message):
    url = 'https://api.line.me/v2/bot/message/reply'
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {LINE_CHANNEL_ACCESS_TOKEN}'
    }
    data = {
        'replyToken': reply_token,
        'messages': [{'type': 'text', 'text': message}]
    }
    response = requests.post(url, json=data, headers=headers)
    if response.status_code != 200:
        print(f"Failed to send reply: {response.status_code}, {response.text}")

# Flask route for LINE webhook
@app.route("/callback", methods=['POST'])
def callback():
    global weight, exercise_days, step, remind_time, ntemperature, humidity

    body = request.get_json()
    if not body or 'events' not in body:
        return 'Invalid request', 400

    for event in body['events']:
        if event['type'] == 'message' and 'text' in event['message']:
            user_message = event['message']['text'].lower()
            reply_token = event['replyToken']
            reply_message = "Sorry, I didn't understand your message."

            if user_message == 'record':
                records = fetch_records()  # Replace with actual function
                if records:
                    reply_message = "Here are your recent water records:\n"
                    for record in records:
                        timestamp, weight, temperature = record
                        reply_message += f"{timestamp}: {weight:.2f} g, {temperature:.2f}C\n"
                else:
                    reply_message = "No records found. Start drinking water to record your intake!"

            elif user_message == 'current temperature':
                object_temp = get_temperature()  # Replace with actual function
                reply_message = f"Drink temperature: {object_temp:.2f} C."
                set_led_based_on_temperature(object_temp)  # Replace with actual function

            elif user_message == 'water intake':
                reply_message = "To calculate your water intake, please send your weight (in kg)."
                step = 'weight'

            elif step == 'weight' and user_message.isdigit():
                weight = int(user_message)
                reply_message = f"Got it! Your weight is {weight} kg. How many days a week do you exercise?"
                step = 'exercise'

            elif step == 'exercise' and user_message.isdigit():
                exercise_days = int(user_message)
                daily_water_intake = weight * 30 + exercise_days * 100
                reply_message = f"Your daily water intake should be {daily_water_intake} ml."
                weight = None
                exercise_days = None

            elif user_message == 'recommend':
                ntemperature, humidity = get_weather_data()  # Replace with actual function
                reply_message = f"Current temperature: {ntemperature}°C, Humidity: {humidity}%.\n"
                reply_message += get_water_intake_recommendation(ntemperature, humidity)  # Replace with actual function

            elif user_message == 'stir':
                reply_message = "How many seconds do you want to mix?"
                step = 'stir'

            elif step == 'stir' and user_message.isdigit():
                duration = int(user_message)
                start_motor(duration)  # Replace with actual function
                reply_message = "Mixer stopped!"
                step = None

            elif user_message == 'remind time':
                reply_message = "Please enter the time in HH:MM format for the reminder."
                step = 'set_remind_time'

            elif step == 'set_remind_time':
                try:
                    datetime.strptime(user_message, "%H:%M")
                    remind_time = user_message
                    reply_message = f"Reminder set for {remind_time}."
                    step = None
                except ValueError:
                    reply_message = "Invalid time format. Please enter the time in HH:MM format."

            send_line_reply(reply_token, reply_message)

    return 'OK'

# LINE Messaging Configuration
def send_line_reply(reply_token, message):
    """Sends a reply message using LINE Messaging API."""
    LINE_CHANNEL_ACCESS_TOKEN = 'YOUR_LINE_CHANNEL_ACCESS_TOKEN'
    url = 'https://api.line.me/v2/bot/message/reply'
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {LINE_CHANNEL_ACCESS_TOKEN}'
    }
    data = {
        'replyToken': reply_token,
        'messages': [{'type': 'text', 'text': message}]
    }

    response = requests.post(url, json=data, headers=headers)
    if response.status_code != 200:
        print(f"Failed to send reply: {response.status_code}, {response.text}")

# Run the Flask app
if __name__ == "__main__":
    app.run(host='0.0.0.0', port=5000)

```
注意: 
* 將CALIBRATION_FACTOR改為，已知重量的變化量/感測器輸出的變化量。
* 將API_KEY改為你在openweathermap上的APIKEY。
* 將LINE_CHANNEL_ACCESS_TOKEN改為你的LINE_CHANNEL_ACCESS_TOKEN。

13. 儲存python檔案並在終端機執行。 
14. 在line官方帳號中輸入訊息，測試各功能。

## 可能改進方向 (Future Improvements)
1. 進一步互動功能
(1). 增加兩顆按鈕，分別記錄飲水或其他飲料的飲用量，以便健康追蹤。
(2). 根據喝水時間或水及其他飲品的比例，推薦理想的飲水量及健康建議，以提升用戶互動性。
(3). 讓杯墊的燈能根據環境調整亮度，或使用Line Bot更改顏色作為照明。
2. 改進使用體驗
	(1). 優化超音波震動頻率及燈光顏色顯示的調整選項。
(2). 整合環境光感測器，以調整 LED 燈亮度。
3. 拓展功能與應用
	(1). 探索將資料儲存於雲端，支援遠端查看飲水紀錄。
(2). 評估市售化潛力，改良設計以符合市場需求。
(3). 進行行為分析，研究人們的飲水習慣與健康的相關性。

## 技術挑戰與解決方案
### 重量感測校正：
初期數據不穩，透過多次取樣平均與誤差修正公式提升穩定度。
### 材質限制：
矽膠片不易黏合，改用紙板製作外殼，並保留未來改進為防水防滑材質。

## 參考資料 (References)
* 以上提及所有官方網站
* https://chatgpt.com/
* https://youtu.be/aNlaj1r7NKc?si=yImRPyllHWDo6eCK
* https://www.icshop.com.tw/products/368030200340
* https://www.youtube.com/watch?v=3YwXp8HjeDY
* https://pinout.xyz/