285 lines
9.3 KiB
JavaScript
285 lines
9.3 KiB
JavaScript
// 采集器: 家庭/设备发现 + 每小时采样 + 电量 + 日聚合 + vendor 补齐编排
|
|
import { prisma } from '../db.js'
|
|
import { checkAuth, getApi, MihomeAuthError } from './client.js'
|
|
import { getSpec } from './specs.js'
|
|
import { dayStart, hourStart, addDays } from './time.js'
|
|
import * as vendorStats from './vendorStats.js'
|
|
|
|
const BACKFILL_DAYS = 45 // 停机/首次接入时, 云端统计补齐窗口
|
|
|
|
const round2 = (n) => (Number.isFinite(n) ? Math.round(n * 100) / 100 : null)
|
|
|
|
// 互斥锁: 同一时刻只允许一个同步在跑 (cron 与手动/启动触发并发时合并)
|
|
let runLock = null
|
|
|
|
function withLock(fn) {
|
|
if (runLock) return runLock
|
|
runLock = Promise.resolve()
|
|
.then(fn)
|
|
.finally(() => {
|
|
runLock = null
|
|
})
|
|
return runLock
|
|
}
|
|
|
|
// ==================== 设备发现 ====================
|
|
|
|
async function syncHomes() {
|
|
const api = getApi()
|
|
const homes = await api.getHomesList()
|
|
for (const h of homes) {
|
|
const homeId = String(h.id)
|
|
await prisma.mihomeHome.upsert({
|
|
where: { homeId },
|
|
create: { homeId, name: h.name || homeId },
|
|
update: { name: h.name || homeId },
|
|
})
|
|
}
|
|
return homes.length
|
|
}
|
|
|
|
async function syncDevices() {
|
|
const api = getApi()
|
|
const list = await api.getDevicesList()
|
|
let kept = 0
|
|
for (const d of list) {
|
|
const spec = getSpec(d.model)
|
|
if (!spec) continue // 非环境设备白名单, 跳过
|
|
const homeId = String(d.home_id || '')
|
|
if (!homeId) continue
|
|
const data = {
|
|
homeId,
|
|
model: d.model,
|
|
name: String(d.name || d.did),
|
|
tempSiid: spec.temp[0],
|
|
tempPiid: spec.temp[1],
|
|
humSiid: spec.hum[0],
|
|
humPiid: spec.hum[1],
|
|
battSiid: spec.batt[0],
|
|
battPiid: spec.batt[1],
|
|
}
|
|
await prisma.mihomeDevice.upsert({
|
|
where: { did: d.did },
|
|
create: { did: d.did, ...data },
|
|
update: data,
|
|
})
|
|
kept++
|
|
}
|
|
return kept
|
|
}
|
|
|
|
async function listEnabledDevices() {
|
|
return prisma.mihomeDevice.findMany({ where: { enabled: true } })
|
|
}
|
|
|
|
// ==================== 采样 ====================
|
|
|
|
// 每小时采样: 批量读温湿度 -> upsert 小时档
|
|
async function sampleHour(devices) {
|
|
const api = getApi()
|
|
const bucket = hourStart()
|
|
const props = []
|
|
for (const d of devices) {
|
|
props.push({ did: d.did, siid: d.tempSiid, piid: d.tempPiid })
|
|
props.push({ did: d.did, siid: d.humSiid, piid: d.humPiid })
|
|
}
|
|
if (props.length === 0) return 0
|
|
|
|
const results = []
|
|
try {
|
|
const res = await api.getDevicesProp(props)
|
|
results.push(...(Array.isArray(res) ? res : [res]))
|
|
} catch (e) {
|
|
// 整体失败: 不写入任何小时档, 缺档留给整点补采/vendor 补齐, 本次直接抛给上层记录
|
|
throw e
|
|
}
|
|
|
|
// did -> { temp, hum }
|
|
const byDevice = new Map(devices.map((d) => [d.did, d]))
|
|
const perDevice = new Map()
|
|
for (const r of results) {
|
|
if (!r || !r.did) continue
|
|
const ok = Number(r.code) === 0 && r.value !== undefined && r.value !== null
|
|
if (!ok) continue
|
|
const d = byDevice.get(r.did)
|
|
if (!d) continue
|
|
const cur = perDevice.get(r.did) || {}
|
|
if (r.siid === d.tempSiid && r.piid === d.tempPiid) {
|
|
cur.temperatureAvg = round2(r.value)
|
|
} else if (r.siid === d.humSiid && r.piid === d.humPiid) {
|
|
cur.humidityAvg = round2(r.value)
|
|
}
|
|
perDevice.set(r.did, cur)
|
|
}
|
|
|
|
let written = 0
|
|
for (const d of devices) {
|
|
const v = perDevice.get(d.did)
|
|
if (!v || (v.temperatureAvg === undefined && v.humidityAvg === undefined)) continue
|
|
await prisma.mihomeEnvHour.upsert({
|
|
where: { deviceId_hour: { deviceId: d.id, hour: bucket } },
|
|
create: {
|
|
deviceId: d.id,
|
|
hour: bucket,
|
|
temperatureAvg: v.temperatureAvg ?? null,
|
|
humidityAvg: v.humidityAvg ?? null,
|
|
raw: null, // 自采
|
|
},
|
|
update: {
|
|
temperatureAvg: v.temperatureAvg ?? null,
|
|
humidityAvg: v.humidityAvg ?? null,
|
|
raw: null,
|
|
},
|
|
})
|
|
await prisma.mihomeDevice.update({
|
|
where: { id: d.id },
|
|
data: { lastSyncAt: new Date() },
|
|
})
|
|
written++
|
|
}
|
|
return written
|
|
}
|
|
|
|
// 电量: 每天读一次 (batteryAt 不在今天就刷新), 标题展示用
|
|
async function refreshBattery(devices) {
|
|
const api = getApi()
|
|
const today = dayStart()
|
|
for (const d of devices) {
|
|
const isFresh = d.batteryAt && d.batteryAt.getTime() >= today.getTime()
|
|
if (isFresh) continue
|
|
try {
|
|
const res = await api.getDevicesProp({ did: d.did, siid: d.battSiid, piid: d.battPiid })
|
|
if (res && Number(res.code) === 0 && typeof res.value === 'number') {
|
|
await prisma.mihomeDevice.update({
|
|
where: { id: d.id },
|
|
data: { battery: Math.round(res.value), batteryAt: new Date() },
|
|
})
|
|
}
|
|
} catch (e) {
|
|
// 电量读不到不报错, 下次再试
|
|
}
|
|
}
|
|
}
|
|
|
|
// ==================== 日聚合 ====================
|
|
|
|
// 按日档重算: 读该上海日小时内档 -> avg/min/max + 最后一次值
|
|
export async function aggregateDay(device, dayDate) {
|
|
const start = dayStart(dayDate.getTime())
|
|
const end = addDays(start, 1)
|
|
const hours = await prisma.mihomeEnvHour.findMany({
|
|
where: { deviceId: device.id, hour: { gte: start, lt: end } },
|
|
orderBy: { hour: 'asc' },
|
|
})
|
|
if (hours.length === 0) return false
|
|
const temps = hours.filter((h) => h.temperatureAvg != null).map((h) => h.temperatureAvg)
|
|
const hums = hours.filter((h) => h.humidityAvg != null).map((h) => h.humidityAvg)
|
|
const last = hours[hours.length - 1]
|
|
const avg = (arr) => (arr.length ? arr.reduce((a, b) => a + b, 0) / arr.length : null)
|
|
const allVendor = hours.every((h) => h.raw && h.raw.vendor === true)
|
|
await prisma.mihomeEnvSample.upsert({
|
|
where: { deviceId_date: { deviceId: device.id, date: start } },
|
|
create: {
|
|
deviceId: device.id,
|
|
date: start,
|
|
temperature: temps.length ? temps[temps.length - 1] : null,
|
|
humidity: hums.length ? hums[hums.length - 1] : null,
|
|
temperatureAvg: round2(avg(temps)),
|
|
humidityAvg: round2(avg(hums)),
|
|
temperatureMin: round2(temps.length ? Math.min(...temps) : null),
|
|
temperatureMax: round2(temps.length ? Math.max(...temps) : null),
|
|
humidityMin: round2(hums.length ? Math.min(...hums) : null),
|
|
humidityMax: round2(hums.length ? Math.max(...hums) : null),
|
|
origin: allVendor ? 'vendor_history' : 'live',
|
|
sampledAt: last.hour,
|
|
raw: { bucketCount: hours.length },
|
|
},
|
|
update: {
|
|
temperature: temps.length ? temps[temps.length - 1] : null,
|
|
humidity: hums.length ? hums[hums.length - 1] : null,
|
|
temperatureAvg: round2(avg(temps)),
|
|
humidityAvg: round2(avg(hums)),
|
|
temperatureMin: round2(temps.length ? Math.min(...temps) : null),
|
|
temperatureMax: round2(temps.length ? Math.max(...temps) : null),
|
|
humidityMin: round2(hums.length ? Math.min(...hums) : null),
|
|
humidityMax: round2(hums.length ? Math.max(...hums) : null),
|
|
origin: allVendor ? 'vendor_history' : 'live',
|
|
sampledAt: last.hour,
|
|
},
|
|
})
|
|
return true
|
|
}
|
|
|
|
// 补聚合缺口日: [since, today) 有小时档却没有日档的日子
|
|
export async function aggregateGapDays(device, since) {
|
|
const today = dayStart()
|
|
const existing = await prisma.mihomeEnvSample.findMany({
|
|
where: { deviceId: device.id, date: { gte: since, lt: today } },
|
|
select: { date: true },
|
|
})
|
|
const have = new Set(existing.map((s) => s.date.getTime()))
|
|
let agg = 0
|
|
for (let t = since.getTime(); t < today.getTime(); t += 86400000) {
|
|
if (!have.has(t)) {
|
|
const ok = await aggregateDay(device, new Date(t))
|
|
if (ok) agg++
|
|
}
|
|
}
|
|
return agg
|
|
}
|
|
|
|
// ==================== 对外入口 ====================
|
|
|
|
// 每次同步: 发现 + 当前小时采样 + 今天/昨天聚合 + 电量 (轻量, 每小时跑)
|
|
// 注意: 单独不包 withLock, 由 syncNow/dailyMaintenance 统一加锁, 避免嵌套自锁
|
|
async function syncLocked() {
|
|
const auth = await checkAuth()
|
|
if (auth.status !== 'ok') throw new MihomeAuthError(auth.lastError || '米家未授权或已过期')
|
|
const began = Date.now()
|
|
const homes = await syncHomes()
|
|
const devices = await syncDevices()
|
|
const kept = await listEnabledDevices()
|
|
const sampled = await sampleHour(kept)
|
|
await refreshBattery(kept)
|
|
// 今天 + 昨天都聚合 (保证 00 点边界不丢)
|
|
const nowMs = Date.now()
|
|
const today = dayStart(nowMs)
|
|
let agg = 0
|
|
for (const d of kept) {
|
|
if (await aggregateDay(d, today)) agg++
|
|
const y = addDays(today, -1)
|
|
if (await aggregateDay(d, y)) agg++
|
|
}
|
|
return {
|
|
ok: true,
|
|
homes,
|
|
devices,
|
|
sampled,
|
|
aggregated: agg,
|
|
ms: Date.now() - began,
|
|
}
|
|
}
|
|
|
|
export function syncNow() {
|
|
return withLock(syncLocked)
|
|
}
|
|
|
|
// 每日维护: 轻量同步 + vendor 补齐历史缺口 (45 天窗口) + 缺口日聚合
|
|
export function dailyMaintenance() {
|
|
return withLock(async () => {
|
|
const r = await syncLocked()
|
|
const auth = await checkAuth()
|
|
if (auth.status !== 'ok') throw new MihomeAuthError(auth.lastError || '米家未授权或已过期')
|
|
const since = addDays(dayStart(), -BACKFILL_DAYS)
|
|
const devices = await listEnabledDevices()
|
|
let hFilled = 0
|
|
let dFilled = 0
|
|
let agg = 0
|
|
for (const d of devices) {
|
|
hFilled += await vendorStats.fillHourGaps(d, since)
|
|
dFilled += await vendorStats.fillDayGaps(d, since)
|
|
agg += await aggregateGapDays(d, since)
|
|
}
|
|
return { ...r, vendorHours: hFilled, vendorDays: dFilled, gapAggregated: agg }
|
|
})
|
|
} |