# 智慧杯墊 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 : 飲水量與溫度偵測
每次杯墊偵測到使用者拿起水杯並放下後，即會自動計算喝水量及偵測溫度。
### 功能2 : 偵測溫度並以光學展示
使用者在LINE Bot輸入"current temperature"後，杯墊會自動偵測飲品溫度，並依照溫度適不適合飲用發出不同顏色的光。
### 功能3 : 智慧攪拌
使用者在LINE Bot輸入"stir"後，會被詢問要攪拌幾秒，回答完成後杯墊會開始攪拌飲品。
### 功能4 : 喝水提醒
使用者在LINE Bot輸入"remind time"後，會被詢問希望於何時被提醒，回答完成後杯墊將會在那個時間傳訊息與發光提醒使用者喝水。
### 功能5 : 每日飲水量計算
使用者在LINE Bot輸入"water intake"後，會被詢問體重及一周運動次數，回答完成後系統會自動計算每日建議飲水量。
### 功能6 : 飲水建議
使用者在LINE Bot輸入"recommend"後，系統將根據氣溫、濕度和當前時間給予使用者飲水建議。
### 功能7 : 紀錄查詢
使用者在LINE Bot輸入"recommend"後，將收到近期喝水量及溫度的紀錄。
## 必備材料與工具 (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)
### 電路圖
### 照片與示意圖
## 步驟教學 (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/
Sign in with Wallet
Connect another wallet