552 lines
16 KiB
Go
552 lines
16 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"flag"
|
||
"fmt"
|
||
"log"
|
||
"math"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"weatherstation/core/internal/config"
|
||
"weatherstation/core/internal/data"
|
||
"weatherstation/core/internal/sms"
|
||
)
|
||
|
||
const (
|
||
providerForecast = "imdroid_mix"
|
||
alertTypeForecast = "forecast_3h_rain"
|
||
alertTypeActual30m = "actual_30m_rain"
|
||
alertTypeNeighbor30 = "actual_30m_neighbor"
|
||
levelRed = "red"
|
||
levelYellow = "yellow"
|
||
forecastRedMM = 8.0
|
||
forecastYellowMM = 4.0
|
||
actualRedMM = 4.0
|
||
actualYellowMM = 2.0
|
||
halfAngleDeg = 90.0
|
||
timeFormatShort = "2006-01-02 15:04"
|
||
defaultCheckTimeout = 20 * time.Second
|
||
)
|
||
|
||
var (
|
||
flagOnce bool
|
||
flagTest bool
|
||
flagTestStIDs string
|
||
flagTestTime string
|
||
flagWhy bool
|
||
)
|
||
|
||
func main() {
|
||
flag.BoolVar(&flagOnce, "once", false, "run checks once immediately (no scheduling)")
|
||
flag.BoolVar(&flagTest, "test", false, "force alerts regardless of thresholds (for dry-run)")
|
||
flag.StringVar(&flagTestStIDs, "station", "", "comma-separated station_id list for test mode")
|
||
flag.StringVar(&flagTestTime, "time", "", "test mode: specify end time (YYYY-MM-DD HH:MM:SS, CST)")
|
||
flag.BoolVar(&flagWhy, "why", false, "in test mode, log reasons when alert not triggered")
|
||
flag.Parse()
|
||
|
||
cfg := config.Load()
|
||
|
||
scli := mustInitSMS(cfg)
|
||
|
||
if flagOnce {
|
||
tick := time.Now()
|
||
runForecastCheck(scli, tick)
|
||
runActualCheck(scli, tick)
|
||
runNeighborActualCheck(scli, tick)
|
||
return
|
||
}
|
||
|
||
go alignAndRunHour10(func(tick time.Time) { runForecastCheck(scli, tick) })
|
||
go alignAndRunHalfHour(func(tick time.Time) { runActualCheck(scli, tick) })
|
||
go alignAndRunHalfHour(func(tick time.Time) { runNeighborActualCheck(scli, tick) })
|
||
|
||
select {}
|
||
}
|
||
|
||
func mustInitSMS(cfg config.Config) *sms.Client {
|
||
cli, err := sms.New(sms.Config{
|
||
AccessKeyID: strings.TrimSpace(cfg.SMS.AccessKeyID),
|
||
AccessKeySecret: strings.TrimSpace(cfg.SMS.AccessKeySecret),
|
||
SignName: strings.TrimSpace(cfg.SMS.SignName),
|
||
TemplateCode: strings.TrimSpace(cfg.SMS.TemplateCode),
|
||
Endpoint: strings.TrimSpace(cfg.SMS.Endpoint),
|
||
})
|
||
if err != nil {
|
||
log.Printf("sms: disabled (%v)", err)
|
||
return nil
|
||
}
|
||
return cli
|
||
}
|
||
|
||
func alignAndRunHour10(fn func(tick time.Time)) {
|
||
now := time.Now()
|
||
base := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 10, 0, 0, now.Location())
|
||
var next time.Time
|
||
if now.After(base) {
|
||
next = base.Add(time.Hour)
|
||
} else {
|
||
next = base
|
||
}
|
||
time.Sleep(time.Until(next))
|
||
for {
|
||
tick := time.Now().Truncate(time.Minute)
|
||
fn(tick)
|
||
time.Sleep(time.Hour)
|
||
}
|
||
}
|
||
|
||
func alignAndRunHalfHour(fn func(tick time.Time)) {
|
||
now := time.Now()
|
||
next := now.Truncate(30 * time.Minute).Add(30 * time.Minute)
|
||
time.Sleep(time.Until(next))
|
||
for {
|
||
tick := time.Now().Truncate(time.Minute)
|
||
fn(tick)
|
||
time.Sleep(30 * time.Minute)
|
||
}
|
||
}
|
||
|
||
func runForecastCheck(scli *sms.Client, tick time.Time) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), defaultCheckTimeout)
|
||
defer cancel()
|
||
|
||
stations := listFixedStations(ctx)
|
||
stations = filterStations(stations, flagTestStIDs)
|
||
if len(stations) == 0 {
|
||
log.Printf("forecast: no stations after filter")
|
||
return
|
||
}
|
||
|
||
recip := loadRecipients(ctx)
|
||
loc := mustShanghai()
|
||
now := tick.In(loc)
|
||
issued := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, loc)
|
||
next1 := issued.Add(time.Hour)
|
||
next3 := issued.Add(3 * time.Hour)
|
||
|
||
for _, st := range stations {
|
||
points, err := data.ForecastRainAtIssued(ctx, st.ID, providerForecast, issued)
|
||
if err != nil {
|
||
log.Printf("forecast: query station=%s err=%v", st.ID, err)
|
||
continue
|
||
}
|
||
var redMax, yellowMax int64
|
||
for _, p := range points {
|
||
if p.ForecastTime.Before(next1) || p.ForecastTime.After(next3) {
|
||
continue
|
||
}
|
||
v := int64(p.RainMMx1000)
|
||
if v >= 8000 {
|
||
if v > redMax {
|
||
redMax = v
|
||
}
|
||
} else if v >= 4000 {
|
||
if v > yellowMax {
|
||
yellowMax = v
|
||
}
|
||
}
|
||
}
|
||
level := ""
|
||
value := 0.0
|
||
threshold := 0.0
|
||
if redMax > 0 {
|
||
level = levelRed
|
||
value = float64(redMax) / 1000.0
|
||
threshold = forecastRedMM
|
||
} else if yellowMax > 0 {
|
||
level = levelYellow
|
||
value = float64(yellowMax) / 1000.0
|
||
threshold = forecastYellowMM
|
||
}
|
||
if level == "" {
|
||
if !flagTest {
|
||
if flagWhy {
|
||
log.Printf("forecast why: station=%s no threshold hit redMax=%d yellowMax=%d", st.ID, redMax, yellowMax)
|
||
}
|
||
continue
|
||
}
|
||
level = levelYellow
|
||
value = forecastYellowMM
|
||
threshold = forecastYellowMM
|
||
}
|
||
if flagTest {
|
||
msg := fmt.Sprintf("【测试】站点%s 强制触发未来3小时降水预警 level=%s", st.Name, levelLabel(level))
|
||
targetPhones := recip.forLevel(level)
|
||
if len(targetPhones) == 0 {
|
||
recordAlert(ctx, alertTypeForecast, st.ID, level, issued, msg, sql.NullString{})
|
||
} else {
|
||
sendToPhones(ctx, scli, st.Name, value, level, issued, targetPhones, msg, alertTypeForecast, st.ID)
|
||
}
|
||
continue
|
||
}
|
||
msg := fmt.Sprintf("站点%s 未来3小时单小时最大降水 %.3fmm，达到%s阈值 %.1fmm，issued_at=%s", st.Name, value, levelLabel(level), threshold, issued.Format(timeFormatShort))
|
||
targetPhones := recip.forLevel(level)
|
||
if len(targetPhones) == 0 {
|
||
recordAlert(ctx, alertTypeForecast, st.ID, level, issued, msg, sql.NullString{})
|
||
continue
|
||
}
|
||
sendToPhones(ctx, scli, st.Name, value, level, issued, targetPhones, msg, alertTypeForecast, st.ID)
|
||
}
|
||
}
|
||
|
||
func runActualCheck(scli *sms.Client, tick time.Time) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), defaultCheckTimeout)
|
||
defer cancel()
|
||
|
||
stations := listFixedStations(ctx)
|
||
stations = filterStations(stations, flagTestStIDs)
|
||
if len(stations) == 0 {
|
||
log.Printf("actual: no stations after filter")
|
||
return
|
||
}
|
||
|
||
recip := loadRecipients(ctx)
|
||
loc := mustShanghai()
|
||
end := tick.In(loc)
|
||
if flagTestTime != "" {
|
||
if t, err := time.ParseInLocation("2006-01-02 15:04:05", flagTestTime, loc); err == nil {
|
||
end = t
|
||
}
|
||
}
|
||
start := end.Add(-30 * time.Minute)
|
||
|
||
for _, st := range stations {
|
||
rain, ok, err := data.SumRainMM(ctx, st.ID, start, end)
|
||
if err != nil {
|
||
log.Printf("actual: sum station=%s err=%v", st.ID, err)
|
||
continue
|
||
}
|
||
if flagWhy {
|
||
log.Printf("actual why: station=%s window=%s~%s rain_sum=%.3f ok=%v", st.ID, start.Format(timeFormatShort), end.Format(timeFormatShort), rain, ok)
|
||
}
|
||
if !ok {
|
||
if flagWhy {
|
||
log.Printf("actual why: station=%s no rain data", st.ID)
|
||
}
|
||
continue
|
||
}
|
||
level := ""
|
||
threshold := 0.0
|
||
if rain >= actualRedMM {
|
||
level = levelRed
|
||
threshold = actualRedMM
|
||
} else if rain >= actualYellowMM {
|
||
level = levelYellow
|
||
threshold = actualYellowMM
|
||
}
|
||
if level == "" {
|
||
if flagWhy {
|
||
log.Printf("actual why: station=%s rain=%.3f below threshold", st.ID, rain)
|
||
}
|
||
if !flagTest {
|
||
continue
|
||
}
|
||
level = levelYellow
|
||
threshold = actualYellowMM
|
||
rain = actualYellowMM
|
||
}
|
||
if flagTest {
|
||
msg := fmt.Sprintf("【测试】站点%s 强制触发30分钟降水预警 level=%s", st.Name, levelLabel(level))
|
||
targetPhones := recip.forLevel(level)
|
||
if len(targetPhones) == 0 {
|
||
recordAlert(ctx, alertTypeActual30m, st.ID, level, end, msg, sql.NullString{})
|
||
} else {
|
||
sendToPhones(ctx, scli, st.Name, rain, level, end, targetPhones, msg, alertTypeActual30m, st.ID)
|
||
}
|
||
continue
|
||
}
|
||
msg := fmt.Sprintf("站点%s 过去30分钟降水 %.3fmm，达到%s阈值 %.1fmm，窗口 %s - %s", st.Name, rain, levelLabel(level), threshold, start.Format(timeFormatShort), end.Format(timeFormatShort))
|
||
targetPhones := recip.forLevel(level)
|
||
if len(targetPhones) == 0 {
|
||
recordAlert(ctx, alertTypeActual30m, st.ID, level, end, msg, sql.NullString{})
|
||
log.Printf("actual: triggered station=%s level=%s rain=%.3f (no phones)", st.ID, level, rain)
|
||
continue
|
||
}
|
||
sendToPhones(ctx, scli, st.Name, rain, level, end, targetPhones, msg, alertTypeActual30m, st.ID)
|
||
log.Printf("actual: triggered station=%s level=%s rain=%.3f phones=%d", st.ID, level, rain, len(targetPhones))
|
||
}
|
||
}
|
||
|
||
func runNeighborActualCheck(scli *sms.Client, tick time.Time) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), defaultCheckTimeout)
|
||
defer cancel()
|
||
|
||
allStations := listFixedStations(ctx)
|
||
centers := filterStations(allStations, flagTestStIDs)
|
||
if len(centers) == 0 {
|
||
if flagWhy {
|
||
log.Printf("neighbor why: no stations after filter")
|
||
}
|
||
return
|
||
}
|
||
|
||
recip := loadRecipients(ctx)
|
||
loc := mustShanghai()
|
||
end := tick.In(loc)
|
||
if flagTestTime != "" {
|
||
if t, err := time.ParseInLocation("2006-01-02 15:04:05", flagTestTime, loc); err == nil {
|
||
end = t
|
||
}
|
||
}
|
||
start := end.Add(-30 * time.Minute)
|
||
|
||
for _, center := range centers {
|
||
if center.Latitude == 0 || center.Longitude == 0 {
|
||
if flagWhy {
|
||
log.Printf("neighbor why: center %s missing lat/lon", center.ID)
|
||
}
|
||
continue
|
||
}
|
||
wind, err := data.RadarWeatherNearest(center.Latitude, center.Longitude, end, 6*time.Hour)
|
||
if err != nil {
|
||
log.Printf("neighbor: wind query failed station=%s: %v", center.ID, err)
|
||
continue
|
||
}
|
||
if wind == nil || !wind.WindDirection.Valid || !wind.WindSpeed.Valid || wind.WindSpeed.Float64 <= 0.01 {
|
||
if flagWhy {
|
||
log.Printf("neighbor why: center %s no wind data", center.ID)
|
||
}
|
||
continue
|
||
}
|
||
dir := wind.WindDirection.Float64
|
||
spd := wind.WindSpeed.Float64
|
||
radius := spd * 3600
|
||
for _, nb := range allStations {
|
||
if nb.ID == center.ID {
|
||
continue
|
||
}
|
||
if nb.Latitude == 0 || nb.Longitude == 0 {
|
||
if flagWhy {
|
||
log.Printf("neighbor why: neighbor %s missing lat/lon", nb.ID)
|
||
}
|
||
continue
|
||
}
|
||
dist := haversine(center.Latitude, center.Longitude, nb.Latitude, nb.Longitude)
|
||
brg := bearingDeg(center.Latitude, center.Longitude, nb.Latitude, nb.Longitude)
|
||
diff := angDiff(brg, dir)
|
||
inSector := dist <= radius && diff <= halfAngleDeg
|
||
if !inSector {
|
||
if flagWhy {
|
||
log.Printf("neighbor why: center=%s neighbor=%s not in sector dist=%.1fm radius=%.1fm bearing=%.1f windFrom=%.1f diff=%.1f half=%.1f",
|
||
center.ID, nb.ID, dist, radius, brg, dir, diff, halfAngleDeg)
|
||
}
|
||
continue
|
||
}
|
||
rain, ok, err := data.SumRainMM(ctx, nb.ID, start, end)
|
||
if err != nil {
|
||
log.Printf("neighbor: sum rain station=%s err=%v", nb.ID, err)
|
||
continue
|
||
}
|
||
if flagWhy {
|
||
log.Printf("neighbor why: center=%s neighbor=%s window=%s~%s rain_sum=%.3f ok=%v", center.ID, nb.ID, start.Format(timeFormatShort), end.Format(timeFormatShort), rain, ok)
|
||
}
|
||
if !ok {
|
||
if flagWhy {
|
||
log.Printf("neighbor why: neighbor %s no rain data", nb.ID)
|
||
}
|
||
continue
|
||
}
|
||
level := ""
|
||
threshold := 0.0
|
||
if rain >= actualRedMM {
|
||
level = levelRed
|
||
threshold = actualRedMM
|
||
} else if rain >= actualYellowMM {
|
||
level = levelYellow
|
||
threshold = actualYellowMM
|
||
}
|
||
if level == "" {
|
||
if flagWhy {
|
||
log.Printf("neighbor why: neighbor %s rain=%.3f below threshold", nb.ID, rain)
|
||
}
|
||
continue
|
||
}
|
||
atype := alertTypeNeighbor30 + "_" + nb.ID
|
||
msg := fmt.Sprintf("站点%s 迎风扇区内站点%s 30分钟降水 %.3fmm，达到%s阈值 %.1fmm，窗口 %s - %s", center.Name, nb.Name, rain, levelLabel(level), threshold, start.Format(timeFormatShort), end.Format(timeFormatShort))
|
||
targetPhones := recip.forLevel(level)
|
||
if len(targetPhones) == 0 {
|
||
recordAlert(ctx, atype, center.ID, level, end, msg, sql.NullString{})
|
||
log.Printf("neighbor: center=%s neighbor=%s level=%s rain=%.3f (no phones)", center.ID, nb.ID, level, rain)
|
||
continue
|
||
}
|
||
sendToPhones(ctx, scli, center.Name, rain, level, end, targetPhones, msg, atype, center.ID)
|
||
log.Printf("neighbor: center=%s neighbor=%s level=%s rain=%.3f phones=%d", center.ID, nb.ID, level, rain, len(targetPhones))
|
||
}
|
||
}
|
||
}
|
||
|
||
func recordAlert(ctx context.Context, alertType, stationID, level string, issuedAt time.Time, message string, phone sql.NullString) {
|
||
_, err := data.InsertAlert(ctx, data.AlertRecord{
|
||
AlertType: alertType,
|
||
StationID: stationID,
|
||
Level: level,
|
||
IssuedAt: issuedAt,
|
||
Message: message,
|
||
SMSPhone: phone,
|
||
})
|
||
if err != nil {
|
||
log.Printf("alert insert failed station=%s type=%s level=%s: %v", stationID, alertType, level, err)
|
||
}
|
||
}
|
||
|
||
type recipients struct {
|
||
red []string
|
||
yel []string
|
||
}
|
||
|
||
func (r recipients) forLevel(level string) []string {
|
||
if level == levelRed {
|
||
return r.red
|
||
}
|
||
if level == levelYellow {
|
||
return r.yel
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func loadRecipients(ctx context.Context) recipients {
|
||
list, err := data.ListEnabledSMSRecipients(ctx)
|
||
if err != nil {
|
||
log.Printf("sms: load recipients failed: %v", err)
|
||
return recipients{}
|
||
}
|
||
var res recipients
|
||
for _, r := range list {
|
||
if r.AlertLevel >= 1 {
|
||
res.red = append(res.red, r.Phone)
|
||
}
|
||
if r.AlertLevel >= 2 {
|
||
res.yel = append(res.yel, r.Phone)
|
||
}
|
||
}
|
||
return res
|
||
}
|
||
|
||
func sendToPhones(ctx context.Context, scli *sms.Client, stationName string, value float64, level string, issuedAt time.Time, phones []string, message string, alertType string, stationID string) {
|
||
if scli == nil {
|
||
return
|
||
}
|
||
name := "：" + stationName + "，"
|
||
content := format3(value) + " mm"
|
||
alertText := "【大礼村】暴雨"
|
||
if level == levelRed {
|
||
alertText += "红色预警"
|
||
} else {
|
||
alertText += "黄色预警"
|
||
}
|
||
for _, ph := range phones {
|
||
if err := scli.Send(ctx, name, content, alertText, "", []string{ph}); err != nil {
|
||
log.Printf("sms: send failed phone=%s station=%s level=%s: %v", ph, stationID, level, err)
|
||
continue
|
||
}
|
||
recordAlert(ctx, alertType, stationID, level, issuedAt, message, sql.NullString{String: ph, Valid: true})
|
||
log.Printf("sms: sent phone=%s station=%s level=%s", ph, stationID, level)
|
||
}
|
||
}
|
||
|
||
func format3(v float64) string {
|
||
s := fmt.Sprintf("%.3f", v)
|
||
s = strings.TrimRight(s, "0")
|
||
s = strings.TrimRight(s, ".")
|
||
if s == "" {
|
||
return "0"
|
||
}
|
||
return s
|
||
}
|
||
|
||
func mustShanghai() *time.Location {
|
||
loc, _ := time.LoadLocation("Asia/Shanghai")
|
||
if loc == nil {
|
||
loc = time.FixedZone("CST", 8*3600)
|
||
}
|
||
return loc
|
||
}
|
||
|
||
func levelLabel(level string) string {
|
||
if level == levelRed {
|
||
return "红色"
|
||
}
|
||
return "黄色"
|
||
}
|
||
|
||
func listFixedStations(ctx context.Context) []data.StationInfo {
|
||
ids := []string{"RS485-002964", "RS485-002A39", "RS485-0029CB"}
|
||
sts, err := data.ListStationsByIDs(ctx, ids)
|
||
if err != nil || len(sts) == 0 {
|
||
var out []data.StationInfo
|
||
for _, id := range ids {
|
||
out = append(out, data.StationInfo{ID: id, Name: id})
|
||
}
|
||
return out
|
||
}
|
||
// 保证顺序按 ids
|
||
order := make(map[string]int)
|
||
for i, id := range ids {
|
||
order[id] = i
|
||
}
|
||
sort.Slice(sts, func(i, j int) bool { return order[sts[i].ID] < order[sts[j].ID] })
|
||
return sts
|
||
}
|
||
|
||
func filterStations(in []data.StationInfo, filter string) []data.StationInfo {
|
||
f := strings.TrimSpace(filter)
|
||
if f == "" {
|
||
return in
|
||
}
|
||
parts := strings.Split(f, ",")
|
||
m := make(map[string]struct{}, len(parts))
|
||
for _, p := range parts {
|
||
p = strings.TrimSpace(p)
|
||
if p != "" {
|
||
m[p] = struct{}{}
|
||
}
|
||
}
|
||
if len(m) == 0 {
|
||
return in
|
||
}
|
||
var out []data.StationInfo
|
||
for _, st := range in {
|
||
if _, ok := m[st.ID]; ok {
|
||
out = append(out, st)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func toRad(d float64) float64 { return d * math.Pi / 180 }
|
||
func toDeg(r float64) float64 { return r * 180 / math.Pi }
|
||
|
||
func haversine(lat1, lon1, lat2, lon2 float64) float64 {
|
||
const R = 6371000.0
|
||
dLat := toRad(lat2 - lat1)
|
||
dLon := toRad(lon2 - lon1)
|
||
a := math.Sin(dLat/2)*math.Sin(dLat/2) + math.Cos(toRad(lat1))*math.Cos(toRad(lat2))*math.Sin(dLon/2)*math.Sin(dLon/2)
|
||
c := 2 * math.Atan2(math.Sqrt(a), math.Sqrt(1-a))
|
||
return R * c
|
||
}
|
||
|
||
func bearingDeg(lat1, lon1, lat2, lon2 float64) float64 {
|
||
φ1 := toRad(lat1)
|
||
φ2 := toRad(lat2)
|
||
Δλ := toRad(lon2 - lon1)
|
||
y := math.Sin(Δλ) * math.Cos(φ2)
|
||
x := math.Cos(φ1)*math.Sin(φ2) - math.Sin(φ1)*math.Cos(φ2)*math.Cos(Δλ)
|
||
brg := toDeg(math.Atan2(y, x))
|
||
if brg < 0 {
|
||
brg += 360
|
||
}
|
||
return brg
|
||
}
|
||
|
||
func angDiff(a, b float64) float64 {
|
||
d := math.Mod(a-b+540, 360) - 180
|
||
if d < 0 {
|
||
d = -d
|
||
}
|
||
return math.Abs(d)
|
||
}
|