goordonchen/m8-smart-charging repository overview

m8-smart-charging/
├── Dockerfile
├── entrypoint.sh                  ← 容器启动入口（环境变量校验 + 进程管理）
├── m8-smart-charging-enhanced.py  ← 主脚本（自动提醒、校准分析）
├── health_check.py                ← 健康检查服务（纯标准库，无额外依赖）
├── requirements.txt
└── logs/                          ← 启动后自动创建

docker run --rm --privileged multiarch/qemu-user-static --reset -p yes

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import sys
import logging
from datetime import datetime, timedelta, timezone
from typing import Optional
from urllib.parse import quote

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

BEIJING_TZ  = timezone(timedelta(hours=8))
NIGHT_HOURS: frozenset[int] = frozenset(range(21, 24)) | frozenset(range(0, 7))

USABLE_CAPACITY: dict[str, float] = {
    "m8_tri_ncm":         51.975,
    "model3_2020_sr_ncm": 51.8,
}
USABLE_CAPACITY["default"] = USABLE_CAPACITY["m8_tri_ncm"]

# 温度策略：List[(最低温阈值℃, 目标充入kWh)]
# 非单调说明：0℃ 以下析锂风险上升需降档；5–25℃ 活性最佳可提升上限。
CHARGE_STRATEGIES: dict[str, list[tuple[float, float]]] = {
    "m8_tri_ncm": [
        (25.0, 36.38),   # ≥25℃ → 70%
        (15.0, 41.58),   # ≥15℃ → 80%
        ( 5.0, 44.18),   # ≥ 5℃ → 85%（活性区间峰值）
        ( 0.0, 31.19),   # ≥ 0℃ → 60%（低温警戒）
        (-99.0, 25.99),  # 极寒  → 50%
    ],
    "model3_2020_sr_ncm": [
        (25.0, 36.26),
        (15.0, 41.44),
        ( 5.0, 44.03),
        ( 0.0, 31.08),
        (-99.0, 25.90),
    ],
}
CHARGE_STRATEGIES["default"] = CHARGE_STRATEGIES["m8_tri_ncm"]

_OFF_PEAK_RAW: list[tuple[tuple[int, ...], str]] = [
    ((3, 4, 5, 6, 10, 11), "22:00 — 08:00"),
    ((7, 8, 9),            "01:00 — 07:00"),
    ((12, 1, 2),           "00:00 — 08:00"),
]
OFF_PEAK_BY_MONTH: dict[int, str] = {
    month: period
    for months, period in _OFF_PEAK_RAW
    for month in months
}

WEATHER_API_URL  = os.getenv("WEATHER_API_URL", "")
WEATHER_API_KEY  = os.getenv("WEATHER_API_KEY", "")
WEATHER_LOCATION = os.getenv("WEATHER_LOCATION", "101270101")
BARK_BASE_URL    = os.getenv("BARK_BASE_URL", "")
BARK_KEY         = os.getenv("BARK_KEY", "")
ICON_URL         = os.getenv("ICON_URL", "")
VEHICLE_MODEL    = os.getenv("VEHICLE_MODEL", "m8_tri_ncm")
# Gist raw 地址，由 charge-location-sync 维护；为空则读 WEATHER_LOCATION 环境变量
LOCATION_URL     = os.getenv("LOCATION_URL", "")

CALIBRATION_WINDOW_START = 13
CALIBRATION_WINDOW_END   = 17
CALIBRATION_TEMP_MIN     = 10.0
CALIBRATION_TEMP_MAX     = 25.0


def get_weather_location() -> tuple[str, str]:
    """
    优先从 Gist 拉取城市信息，格式：城市ID|城市名。
    失败或未配置时 fallback 到 WEATHER_LOCATION 环境变量，城市名返回空字符串。
    """
    if LOCATION_URL:
        try:
            r = requests.get(LOCATION_URL, timeout=5)
            r.raise_for_status()
            raw = r.text.strip()
            if raw:
                parts     = raw.split("|", 1)
                city_id   = parts[0].strip()
                city_name = parts[1].strip() if len(parts) > 1 else ""
                logger.info("从 Gist 读取城市：%s（%s）", city_name or "未知", city_id)
                return city_id, city_name
            logger.warning("Gist 内容为空，使用环境变量城市 ID")
        except Exception as exc:
            logger.warning("Gist 拉取失败，使用环境变量城市 ID：%s", exc)
    logger.info("使用环境变量城市 ID：%s", WEATHER_LOCATION)
    return WEATHER_LOCATION, ""


def validate_env() -> None:
    missing = [v for v in ["WEATHER_API_KEY", "BARK_BASE_URL", "BARK_KEY"] if not os.getenv(v)]
    if missing:
        logger.error("必填环境变量未设置：%s", ", ".join(missing))
        sys.exit(1)


def fetch_hourly_weather(location: str) -> list[dict]:
    url = f"{WEATHER_API_URL}?location={location}&key={WEATHER_API_KEY}&gzip=n"
    try:
        res = requests.get(url, timeout=10)
        res.raise_for_status()
        return res.json().get("hourly", [])
    except Exception as exc:
        logger.error("天气接口请求失败：%s", exc)
        return []


def extract_night_min_temp(hourly: list[dict]) -> Optional[float]:
    temps: list[float] = []
    for entry in hourly:
        fx_time, temp_str = entry.get("fxTime", ""), entry.get("temp")
        if not fx_time or temp_str is None:
            continue
        try:
            if datetime.fromisoformat(fx_time).hour in NIGHT_HOURS:
                temps.append(float(temp_str))
        except (ValueError, TypeError):
            pass
    return min(temps) if temps else None


def suggest_charge_limit(temp: float, model: str) -> str:
    strategy = CHARGE_STRATEGIES.get(model, CHARGE_STRATEGIES["default"])
    usable   = USABLE_CAPACITY.get(model, USABLE_CAPACITY["default"])
    for threshold, target_kwh in strategy:
        if temp >= threshold:
            return f"建议充电至 {round(target_kwh / usable * 100)}%（约 {target_kwh:.1f} kWh）"
    _, last_kwh = strategy[-1]
    return f"建议充电至 {round(last_kwh / usable * 100)}%（约 {last_kwh:.1f} kWh）"


def build_calibration_advice(temp: Optional[float]) -> str:
    if temp is None:
        return (
            "📆 本周为月度校准期，但天气数据获取失败。\n"
            "请手动确认室温是否适合执行 BMS 校准\n"
            "（建议在 10–25℃ 环境下放电至 20% 后充满至 100%）"
        )
    if CALIBRATION_TEMP_MIN <= temp <= CALIBRATION_TEMP_MAX:
        return (
            f"📆 本周为月度校准期，今晚最低气温约 {temp:.1f}℃，温度适宜。\n"
            "🔁 建议执行 BMS 校准：放电至 20% 以下，再充满至 100%"
        )
    reason = "过低" if temp < CALIBRATION_TEMP_MIN else "过高"
    return (
        f"📆 本周为月度校准期，但今晚最低气温约 {temp:.1f}℃（温度{reason}）。\n"
        "⏸️ 暂不建议执行深度校准，本次按常规充电策略执行"
    )


def get_off_peak_period(month: int) -> str:
    return OFF_PEAK_BY_MONTH.get(month, "未知时段（请核查电价政策）")


def push_bark(title: str, body: str) -> None:
    url = f"{BARK_BASE_URL}/{BARK_KEY}/{quote(title.encode('utf-8'))}/{quote(body.encode('utf-8'))}"
    params: dict = {"isArchive": "1", "ttl": "86400"}
    if ICON_URL:
        params["icon"] = ICON_URL
    logger.info("推送目标：%s", url)
    try:
        r = requests.get(url, params=params, timeout=5)
        logger.info("推送结果：%s %s", r.status_code, r.text)
        if r.status_code != 200:
            raise RuntimeError(f"Bark 返回非 200 状态码：{r.status_code}")
    except requests.RequestException as exc:
        raise RuntimeError(f"Bark 网络请求失败：{exc}") from exc


def main() -> None:
    validate_env()
    now = datetime.now(BEIJING_TZ)
    in_calibration_window = CALIBRATION_WINDOW_START <= now.day <= CALIBRATION_WINDOW_END

    location, city_name = get_weather_location()
    hourly = fetch_hourly_weather(location)
    temp   = extract_night_min_temp(hourly)
    logger.info("夜间最低温：%s℃", temp if temp is not None else "获取失败")

    off_peak  = get_off_peak_period(now.month)
    title     = "🔋 今日充电提醒"
    city_line = f"📍 {city_name}\n" if city_name else ""

    if in_calibration_window:
        calibration_text = build_calibration_advice(temp)
        if temp is not None and not (CALIBRATION_TEMP_MIN <= temp <= CALIBRATION_TEMP_MAX):
            body = (
                f"{city_line}{calibration_text}\n"
                f"⚡ {suggest_charge_limit(temp, VEHICLE_MODEL)}\n"
                f"🕰️ 低谷充电时段：{off_peak}"
            )
        else:
            body = (
                f"{city_line}{calibration_text}\n"
                f"🕰️ 低谷充电时段：{off_peak}\n"
                "⚠️ 校准期间请在低谷时段开始充电，避免满电长时间搁置"
            )
    elif temp is not None:
        body = (
            f"{city_line}"
            f"🌡️ 今晚最低气温约 {temp:.1f}℃\n"
            f"⚡ {suggest_charge_limit(temp, VEHICLE_MODEL)}\n"
            f"🕰️ 低谷充电时段：{off_peak}"
        )
    else:
        body = (
            f"{city_line}"
            "⚠️ 天气数据获取失败，请手动设定充电上限\n"
            f"🕰️ 低谷充电时段：{off_peak}"
        )

    logger.info("推送内容：\n%s", body)
    try:
        push_bark(title, body)
    except RuntimeError as exc:
        logger.error("推送失败：%s", exc)
        sys.exit(1)


if __name__ == "__main__":
    main()

from http.server import HTTPServer, BaseHTTPRequestHandler

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"OK")
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, *_):
        pass

if __name__ == "__main__":
    HTTPServer(("0.0.0.0", 8080), Handler).serve_forever()

#!/bin/sh
set -e

# 必填变量缺失时拒绝启动
for var in WEATHER_API_KEY BARK_BASE_URL BARK_KEY; do
    eval val=\$$var
    if [ -z "$val" ]; then
        echo "[ERROR] 必填环境变量 $var 未设置，容器退出" >&2
        exit 1
    fi
done

mkdir -p /app/logs
touch /app/logs/charging.log

# 写入 cron 任务
echo "${CRON_SCHEDULE:-30 18 * * *} python /app/m8-smart-charging-enhanced.py >> /app/logs/charging.log 2>&1" \
    > /etc/crontabs/root

# 后台启动健康检查
python /app/health_check.py &

# 后台启动 crond
crond -f -l 8 &

# tail -f 作为前台进程，docker logs 读取其输出
# 捕获 SIGTERM 优雅退出
trap 'kill $(jobs -p) 2>/dev/null; exit 0' TERM INT
tail -f /app/logs/charging.log

FROM python:3.12-alpine

RUN apk add --no-cache tzdata busybox-suid curl \
    && cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \
    && echo "Asia/Shanghai" > /etc/timezone

WORKDIR /app

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

COPY m8-smart-charging-enhanced.py health_check.py entrypoint.sh ./
RUN mkdir -p /app/logs \
    && touch /app/logs/charging.log \
    && chmod +x /app/entrypoint.sh

ENTRYPOINT ["/app/entrypoint.sh"]

HEALTHCHECK --interval=30s --timeout=5s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8080/health || exit 1

requests==2.32.3

services:
  m8-smart-charging:
    image: goordonchen/m8-smart-charging:latest
    container_name: m8-smart-charging
    restart: unless-stopped
    environment:
      TZ:               Asia/Shanghai
      CRON_SCHEDULE:    "30 18 * * *"
      WEATHER_API_URL:  "https://xxxxx.qweatherapi.com/v7/weather/24h"
      WEATHER_API_KEY:  "你的API_KEY"
      WEATHER_LOCATION: "101270101"
      BARK_BASE_URL:    "https://api.day.app"
      BARK_KEY:         "你的BARK_KEY"
      ICON_URL:         "https://fgw2world.site/tuchuang/AITO_logo.png"
      VEHICLE_MODEL:    "m8_tri_ncm"
      LOCATION_URL:     ""   # Gist raw 地址，配合 charge-location-sync 跨服务器更新城市
    volumes:
      - ./logs:/app/logs
    logging:
      driver: json-file
      options:
        max-size: "10m"
        max-file: "3"

# 注册 QEMU（仅首次需要）
docker run --rm --privileged multiarch/qemu-user-static --reset -p yes

# 创建多平台 builder（仅首次需要）
docker buildx create --name multiarch --driver docker-container --bootstrap --use

# 构建并推送，同时打两个标签
docker buildx build \
  --platform linux/amd64,linux/arm64 \
  -t goordonchen/m8-smart-charging:1.0.1 \
  -t goordonchen/m8-smart-charging:latest \
  --push \
  .

docker-compose up -d

docker run -d \
  --name m8-smart-charging \
  --restart unless-stopped \
  -e TZ=Asia/Shanghai \
  -e CRON_SCHEDULE="30 18 * * *" \
  -e WEATHER_API_URL="https://xxxxx.qweatherapi.com/v7/weather/24h" \
  -e WEATHER_API_KEY="你的API_KEY" \
  -e WEATHER_LOCATION="101270101" \
  -e BARK_BASE_URL="https://api.day.app" \
  -e BARK_KEY="你的BARK_KEY" \
  -e ICON_URL="https://example.com/icon.png" \
  -e VEHICLE_MODEL="m8_tri_ncm" \
  -e LOCATION_URL="" \
  -v "$(pwd)/logs:/app/logs" \
  goordonchen/m8-smart-charging:latest

# 实时日志
docker logs -f m8-smart-charging

# 查看日志文件
tail -f logs/charging.log

# 手动触发一次推送
docker exec -it m8-smart-charging python /app/m8-smart-charging-enhanced.py

# 查看当前 cron 任务
docker exec -it m8-smart-charging crontab -l