kma 구현

.gitignore

@@ -3,3 +3,4 @@
 /.vscode/launch.json
 /.env
 /openweather
+/kma

cmd/kma/handler.go

@@ -0,0 +1,251 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/neatflowcv/seven-skies/internal/pkg/broker"
+	"github.com/neatflowcv/seven-skies/internal/pkg/domain"
+	"github.com/neatflowcv/seven-skies/internal/pkg/kma"
+)
+
+type Handler struct {
+	broker broker.Broker
+	key    string
+	nx     int
+	ny     int
+}
+
+func NewHandler(b broker.Broker, key string, nx int, ny int) *Handler {
+	return &Handler{
+		broker: b,
+		key:    key,
+		nx:     nx,
+		ny:     ny,
+	}
+}
+
+type TemporalData struct {
+	Temperature   string
+	Sky           string
+	Precipitation string
+	ForecastDate  string
+	TargetDate    string
+}
+
+func buildTemporalData(resp *kma.ForecastResponse) []TemporalData {
+	dates := map[string]*TemporalData{}
+	for _, item := range resp.Response.Body.Items.Item {
+		data, ok := dates[item.FcstDate+item.FcstTime]
+		if !ok {
+			data = &TemporalData{
+				ForecastDate:  item.BaseDate + item.BaseTime,
+				TargetDate:    item.FcstDate + item.FcstTime,
+				Temperature:   "",
+				Sky:           "",
+				Precipitation: "",
+			}
+		}
+
+		switch item.Category {
+		case "T1H", "TMP": // 온도
+			data.Temperature = item.FcstValue
+		case "SKY": // 하늘상태
+			data.Sky = item.FcstValue
+		case "PTY": // 강수형태
+			data.Precipitation = item.FcstValue
+		}
+
+		dates[item.FcstDate+item.FcstTime] = data
+	}
+
+	var ret []TemporalData
+	for _, data := range dates {
+		ret = append(ret, *data)
+	}
+
+	return ret
+}
+
+func parseAndValidateTemporalData(data TemporalData) (*domain.WeatherEvent, bool) {
+	targetTime, err := parseKMAForecastDate(data.TargetDate)
+	if err != nil {
+		log.Println("error in parse forecast datetime", err)
+
+		return nil, false
+	}
+
+	forecastTime, err := parseKMAForecastDate(data.ForecastDate)
+	if err != nil {
+		log.Println("error in parse forecast datetime", err)
+
+		return nil, false
+	}
+
+	temperature, err := strconv.ParseInt(data.Temperature, 10, 64)
+	if err != nil {
+		log.Println("error in parse temperature", err)
+
+		return nil, false
+	}
+
+	condition := decideCondition(&data)
+
+	// 값 검증: 비어있는 값이 있으면 건너뛰기
+	if forecastTime.IsZero() {
+		log.Printf("skip weather event: forecastTime is empty (targetDate: %v)", targetTime)
+
+		return nil, false
+	}
+
+	if targetTime.IsZero() {
+		log.Printf("skip weather event: targetTime is empty (forecastDate: %v)", forecastTime)
+
+		return nil, false
+	}
+
+	if condition == "" {
+		log.Printf("skip weather event: condition is empty (forecastDate: %v, targetDate: %v)", forecastTime, targetTime)
+
+		return nil, false
+	}
+
+	if temperature == 0 {
+		log.Printf("skip weather event: temperature is empty (forecastDate: %v, targetDate: %v)", forecastTime, targetTime)
+
+		return nil, false
+	}
+
+	return &domain.WeatherEvent{
+		Source:       domain.WeatherSourceKMA,
+		ForecastDate: forecastTime,
+		TargetDate:   targetTime,
+		Condition:    condition,
+		Temperature: domain.Temperature{
+			Value: domain.Celsius(temperature),
+		},
+	}, true
+}
+
+func convertTemporalDataToWeatherEvents(datas []TemporalData) []*domain.WeatherEvent {
+	var weatherEvents []*domain.WeatherEvent
+
+	for _, data := range datas {
+		weatherEvent, ok := parseAndValidateTemporalData(data)
+		if !ok {
+			continue
+		}
+
+		weatherEvents = append(weatherEvents, weatherEvent)
+	}
+
+	return weatherEvents
+}
+
+// HandleUltraShort는 초단기 예보를 조회해 이벤트로 발행한다.
+func (h *Handler) HandleUltraShort(ctx context.Context) error {
+	log.Println("get ultra short forecast")
+
+	now := time.Now()
+
+	resp, err := kma.GetUltraShortForecast(ctx, h.key, now, h.nx, h.ny)
+	if err != nil {
+		return fmt.Errorf("error in get ultra short forecast: %w", err)
+	}
+
+	datas := buildTemporalData(resp)
+	weatherEvents := convertTemporalDataToWeatherEvents(datas)
+
+	for _, weatherEvent := range weatherEvents {
+		message, err := json.Marshal(weatherEvent)
+		if err != nil {
+			log.Println("error in marshal weather event", err)
+
+			continue
+		}
+
+		err = h.broker.Publish(ctx, "SEVEN_SKIES_SUBJECT.KMA", message)
+		if err != nil {
+			log.Println("error in publish weather event", err)
+
+			continue
+		}
+	}
+
+	return nil
+}
+
+func decideCondition(data *TemporalData) domain.WeatherCondition {
+	if data.Precipitation == "0" && data.Sky == "1" {
+		return domain.WeatherConditionClear
+	}
+
+	switch data.Precipitation {
+	case "1", "5":
+		return domain.WeatherConditionRain
+	case "2", "6":
+		return domain.WeatherConditionRainSnow
+	case "3", "7":
+		return domain.WeatherConditionSnow
+	}
+
+	switch data.Sky {
+	case "3", "4":
+		return domain.WeatherConditionCloudy
+	}
+
+	return domain.WeatherConditionUnknown
+}
+
+// HandleShortTerm는 단기 예보를 조회해 이벤트로 발행한다.
+func (h *Handler) HandleShortTerm(ctx context.Context) error {
+	log.Println("get short term forecast")
+
+	now := time.Now()
+
+	resp, err := kma.GetShortTermForecast(ctx, h.key, now, h.nx, h.ny)
+	if err != nil {
+		return fmt.Errorf("error in get short term forecast: %w", err)
+	}
+
+	datas := buildTemporalData(resp)
+	weatherEvents := convertTemporalDataToWeatherEvents(datas)
+
+	for _, weatherEvent := range weatherEvents {
+		message, err := json.Marshal(weatherEvent)
+		if err != nil {
+			log.Println("error in marshal weather event", err)
+
+			continue
+		}
+
+		err = h.broker.Publish(ctx, "SEVEN_SKIES_SUBJECT.KMA", message)
+		if err != nil {
+			log.Println("error in publish weather event", err)
+
+			continue
+		}
+	}
+
+	return nil
+}
+
+func parseKMAForecastDate(dateStr string) (time.Time, error) {
+	layout := "200601021504"
+
+	loc, err := time.LoadLocation("Asia/Seoul")
+	if err != nil {
+		panic(err)
+	}
+
+	t, err := time.ParseInLocation(layout, dateStr, loc)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("error in parse datetime: %w", err)
+	}
+
+	return t, nil
+}

cmd/kma/main.go

@@ -0,0 +1,191 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"runtime/debug"
+	"strconv"
+
+	"github.com/go-co-op/gocron/v2"
+	"github.com/joho/godotenv"
+	"github.com/neatflowcv/seven-skies/internal/pkg/broker/nats"
+)
+
+func version() string {
+	info, ok := debug.ReadBuildInfo()
+	if !ok {
+		return "unknown"
+	}
+
+	return info.Main.Version
+}
+
+type Config struct {
+	Key     string
+	NX      int
+	NY      int
+	NATSURL string
+}
+
+func NewConfig() *Config {
+	key := os.Getenv("KMA_API_KEY")
+	if key == "" {
+		log.Panic("KMA_API_KEY is not set")
+	}
+
+	nx := os.Getenv("KMA_NX")
+	if nx == "" {
+		log.Panic("KMA_NX is not set")
+	}
+
+	ny := os.Getenv("KMA_NY")
+	if ny == "" {
+		log.Panic("KMA_NY is not set")
+	}
+
+	NATS_URL := os.Getenv("NATS_URL")
+	if NATS_URL == "" {
+		log.Panic("NATS_URL is not set")
+	}
+
+	nxInt, err := strconv.Atoi(nx)
+	if err != nil {
+		log.Panic("error in parse nx", err)
+	}
+
+	nyInt, err := strconv.Atoi(ny)
+	if err != nil {
+		log.Panic("error in parse ny", err)
+	}
+
+	return &Config{
+		Key:     key,
+		NX:      nxInt,
+		NY:      nyInt,
+		NATSURL: NATS_URL,
+	}
+}
+
+func setupBroker(ctx context.Context, natsURL string) (*nats.Broker, error) {
+	broker, err := nats.NewBroker(ctx, natsURL, "SEVEN_SKIES_STREAM", []string{"SEVEN_SKIES_SUBJECT.>"})
+	if err != nil {
+		return nil, fmt.Errorf("error in create broker: %w", err)
+	}
+
+	return broker, nil
+}
+
+//nolint:ireturn // gocron.Scheduler is an interface by design
+func setupScheduler() (gocron.Scheduler, error) {
+	scheduler, err := gocron.NewScheduler()
+	if err != nil {
+		return nil, fmt.Errorf("error in create scheduler: %w", err)
+	}
+
+	return scheduler, nil
+}
+
+func runInitialTasks(ctx context.Context, handler *Handler) error {
+	err := handler.HandleUltraShort(ctx)
+	if err != nil {
+		return err
+	}
+
+	err = handler.HandleShortTerm(ctx)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+//nolint:ireturn // gocron.Job is an interface by design
+func setupJobs(scheduler gocron.Scheduler, handler *Handler) (gocron.Job, gocron.Job, error) {
+	ultraJob, err := scheduler.NewJob(
+		gocron.CronJob("50 * * * *", false),
+		gocron.NewTask(handler.HandleUltraShort),
+	)
+	if err != nil {
+		return nil, nil, fmt.Errorf("error in create ultra short job: %w", err)
+	}
+
+	shortJob, err := scheduler.NewJob(
+		gocron.CronJob("15 2,5,8,11,14,17,20,23 * * *", false),
+		gocron.NewTask(handler.HandleShortTerm),
+	)
+	if err != nil {
+		return nil, nil, fmt.Errorf("error in create short term job: %w", err)
+	}
+
+	return ultraJob, shortJob, nil
+}
+
+func logNextRuns(ultraJob, shortJob gocron.Job) error {
+	ultraNext, err := ultraJob.NextRun()
+	if err != nil {
+		return fmt.Errorf("error in get ultra short next run: %w", err)
+	}
+
+	shortNext, err := shortJob.NextRun()
+	if err != nil {
+		return fmt.Errorf("error in get short term next run: %w", err)
+	}
+
+	log.Println("ultra short next run", ultraNext)
+	log.Println("short term next run", shortNext)
+
+	return nil
+}
+
+func main() {
+	log.Println("version", version())
+
+	err := godotenv.Load()
+	if err == nil {
+		log.Println("env loaded")
+	}
+
+	config := NewConfig()
+	ctx := context.Background()
+
+	broker, err := setupBroker(ctx, config.NATSURL)
+	if err != nil {
+		log.Panic("error in create broker", err)
+	}
+	defer broker.Close()
+
+	handler := NewHandler(broker, config.Key, config.NX, config.NY)
+
+	scheduler, err := setupScheduler()
+	if err != nil {
+		log.Panic("error in create scheduler", err)
+	}
+
+	defer func() {
+		err := scheduler.Shutdown()
+		if err != nil {
+			log.Println("error in shutdown scheduler", err)
+		}
+	}()
+
+	err = runInitialTasks(ctx, handler)
+	if err != nil {
+		log.Panic("error in initial tasks", err)
+	}
+
+	ultraJob, shortJob, err := setupJobs(scheduler, handler)
+	if err != nil {
+		log.Panic("error in create jobs", err)
+	}
+
+	scheduler.Start()
+
+	err = logNextRuns(ultraJob, shortJob)
+	if err != nil {
+		log.Panic("error in get next runs", err)
+	}
+
+	select {}
+}

internal/pkg/kma/errors.go

@@ -0,0 +1,7 @@
+package kma
+
+import "errors"
+
+var (
+	ErrForecast = errors.New("error in get forecast")
+)

internal/pkg/kma/forecaster.go

@@ -0,0 +1,179 @@
+package kma
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"strconv"
+	"time"
+
+	"resty.dev/v3"
+)
+
+const (
+	UltraShortForecastInterval       = time.Hour
+	UltraShortForecastAvailableAfter = 15 * time.Minute
+
+	ShortTermForecastInterval       = 3 * time.Hour
+	ShortTermForecastAvailableAfter = 10 * time.Minute
+)
+
+func GetUltraShortForecast(ctx context.Context, key string, base time.Time, nx int, ny int) (*ForecastResponse, error) {
+	// 초단기 예보는 0:30, 1:30, ..., 11:30 단위로 제공된다.
+	// 요청 시각(`base`)을 기준으로, 그보다 크지 않으면서 가장 가까운 기준 시각을 선택한다.
+	selectedBase := selectUltraShortBase(base)
+
+	client := resty.New()
+
+	defer func() {
+		err := client.Close()
+		if err != nil {
+			log.Println("error in close client", err)
+		}
+	}()
+
+	resp, err := client.R().
+		SetContext(ctx).
+		SetQueryParam("pageNo", "1").
+		SetQueryParam("numOfRows", "1000").
+		SetQueryParam("dataType", "JSON").
+		SetQueryParam("base_date", selectedBase.Format("20060102")).
+		SetQueryParam("base_time", selectedBase.Format("1504")).
+		SetQueryParam("nx", strconv.Itoa(nx)).
+		SetQueryParam("ny", strconv.Itoa(ny)).
+		SetQueryParam("authKey", key).
+		SetResult(&ForecastResponse{}). //nolint:exhaustruct
+		Get("https://apihub.kma.go.kr/api/typ02/openApi/VilageFcstInfoService_2.0/getUltraSrtFcst")
+	if err != nil {
+		return nil, fmt.Errorf("error in get forecast: %w", err)
+	}
+
+	if resp.IsError() {
+		return nil, fmt.Errorf("%w: %s, %s", ErrForecast, resp.Status(), resp.String())
+	}
+
+	return resp.Result().(*ForecastResponse), nil //nolint:forcetypeassert
+}
+
+func GetShortTermForecast(ctx context.Context, key string, base time.Time, nx int, ny int) (*ForecastResponse, error) {
+	selectedBase := selectShortTermBase(base)
+
+	client := resty.New()
+
+	defer func() {
+		err := client.Close()
+		if err != nil {
+			log.Println("error in close client", err)
+		}
+	}()
+
+	resp, err := client.R().SetDebug(true).
+		SetContext(ctx).
+		SetQueryParam("pageNo", "1").
+		SetQueryParam("numOfRows", "1000").
+		SetQueryParam("dataType", "JSON").
+		SetQueryParam("base_date", selectedBase.Format("20060102")).
+		SetQueryParam("base_time", selectedBase.Format("1504")).
+		SetQueryParam("nx", strconv.Itoa(nx)).
+		SetQueryParam("ny", strconv.Itoa(ny)).
+		SetQueryParam("authKey", key).
+		SetResult(&ForecastResponse{}). //nolint:exhaustruct
+		Get("https://apihub.kma.go.kr/api/typ02/openApi/VilageFcstInfoService_2.0/getVilageFcst")
+	if err != nil {
+		return nil, fmt.Errorf("error in get forecast: %w", err)
+	}
+
+	if resp.IsError() {
+		return nil, fmt.Errorf("%w: %s, %s", ErrForecast, resp.Status(), resp.String())
+	}
+
+	return resp.Result().(*ForecastResponse), nil //nolint:forcetypeassert
+}
+
+// 초단기 예보 기준 시각(0:30, 1:30, ..., 11:30) 중에서,
+// 주어진 시각 t 기준으로 t 보다 크지 않으면서 가장 가까운 시각을 반환한다.
+func selectUltraShortBase(base time.Time) time.Time {
+	// 초단기 예보:
+	// - 기준 시각 간격: 1시간
+	// - 조회 가능 시각: 기준 시각 + 15분
+	// - 후보 시작 시각: 이전날 23:30
+	const (
+		startHour   = 23
+		startMinute = 30
+	)
+
+	return selectBase(
+		base,
+		UltraShortForecastInterval,
+		UltraShortForecastAvailableAfter,
+		startHour,
+		startMinute,
+	)
+}
+
+// 단기 예보 기준 시각(02, 05, 08, 11, 14, 17, 20, 23) 중에서,
+// 주어진 시각 t 기준으로, t 시각에서 조회 가능한 가장 최근 기준 시각을 반환한다.
+// 각 기준 시각은 기준 시각 + 10분 이후부터 조회 가능하다.
+func selectShortTermBase(base time.Time) time.Time {
+	// 단기 예보:
+	// - 기준 시각 간격: 3시간
+	// - 조회 가능 시각: 기준 시각 + 10분
+	// - 후보 시작 시각: 이전날 23:00
+	const (
+		startHour   = 23
+		startMinute = 0
+	)
+
+	return selectBase(
+		base,
+		ShortTermForecastInterval,
+		ShortTermForecastAvailableAfter,
+		startHour,
+		startMinute,
+	)
+}
+
+func selectBase(base time.Time, interval, availableAfter time.Duration, startHour, startMinute int) time.Time {
+	// KMA API 는 한국 표준시(Asia/Seoul)를 기준으로 동작하므로,
+	// base 가 다른 타임존(예: UTC)인 경우 한국 시간으로 변환해서 계산한다.
+	loc, err := time.LoadLocation("Asia/Seoul")
+	if err != nil {
+		// 로케일 정보를 불러오지 못하면, 기존 base 의 로케이션을 그대로 사용한다.
+		loc = base.Location()
+	}
+
+	baseInKST := base.In(loc)
+
+	prevDay := baseInKST.AddDate(0, 0, -1)
+
+	start := time.Date(prevDay.Year(), prevDay.Month(), prevDay.Day(), startHour, startMinute, 0, 0, loc)
+	end := baseInKST
+
+	var candidates []time.Time
+
+	for t := start; !t.After(end); t = t.Add(interval) {
+		candidates = append(candidates, t)
+	}
+
+	return selectBaseFromCandidates(baseInKST, candidates, availableAfter)
+}
+
+// 공통 기준 시각 선택 로직:
+// - candidates: 오름차순(과거 -> 미래) 기준 시각 목록
+// - 각 기준 시각은 availableAfter 이후부터 조회 가능
+// - base 시각 기준으로, 조회 가능한 가장 최근 기준 시각을 반환한다.
+func selectBaseFromCandidates(base time.Time, candidates []time.Time, availableAfter time.Duration) time.Time {
+	// 가장 최근 기준 시각을 찾기 위해 역순으로 순회한다.
+	for i := len(candidates) - 1; i >= 0; i-- {
+		candidate := candidates[i]
+		availableAt := candidate.Add(availableAfter)
+
+		if base.Before(availableAt) {
+			continue
+		}
+
+		return candidate
+	}
+
+	panic("no available base time found")
+}

internal/pkg/kma/model.go

@@ -0,0 +1,38 @@
+package kma
+
+type ForecastResponse struct {
+	Response Response `json:"response,omitzero"`
+}
+
+type Header struct {
+	ResultCode string `json:"resultCode,omitempty"`
+	ResultMsg  string `json:"resultMsg,omitempty"`
+}
+
+type Item struct {
+	BaseDate  string `json:"baseDate,omitempty"`
+	BaseTime  string `json:"baseTime,omitempty"`
+	Category  string `json:"category,omitempty"`
+	FcstDate  string `json:"fcstDate,omitempty"`
+	FcstTime  string `json:"fcstTime,omitempty"`
+	FcstValue string `json:"fcstValue,omitempty"`
+	Nx        int    `json:"nx,omitempty"`
+	Ny        int    `json:"ny,omitempty"`
+}
+
+type Items struct {
+	Item []Item `json:"item,omitempty"`
+}
+
+type Body struct {
+	DataType   string `json:"dataType,omitempty"`
+	Items      Items  `json:"items,omitzero"`
+	PageNo     int    `json:"pageNo,omitempty"`
+	NumOfRows  int    `json:"numOfRows,omitempty"`
+	TotalCount int    `json:"totalCount,omitempty"`
+}
+
+type Response struct {
+	Header Header `json:"header,omitzero"`
+	Body   Body   `json:"body,omitzero"`
+}