// FILE: ./internal/sources/openweather/observation.go package openweather import ( "context" "encoding/json" "fmt" "net/url" "strings" "time" "gitea.maximumdirect.net/ejr/feedkit/config" "gitea.maximumdirect.net/ejr/feedkit/event" "gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/common" "gitea.maximumdirect.net/ejr/weatherfeeder/internal/standards" ) // ObservationSource polls the OpenWeatherMap "Current weather" endpoint and emits a RAW observation Event. // // IMPORTANT UNIT POLICY (weatherfeeder convention): // OpenWeather changes units based on the `units` query parameter but does NOT include the unit // system in the response body. To keep normalization deterministic, this driver *requires* // `units=metric`. If absent (or non-metric), the driver returns an error. type ObservationSource struct { http *common.HTTPSource } func NewObservationSource(cfg config.SourceConfig) (*ObservationSource, error) { const driver = "openweather_observation" hs, err := common.NewHTTPSource(driver, cfg, "application/json") if err != nil { return nil, err } if err := requireMetricUnits(hs.URL); err != nil { return nil, fmt.Errorf("%s %q: %w", hs.Driver, hs.Name, err) } return &ObservationSource{http: hs}, nil } func (s *ObservationSource) Name() string { return s.http.Name } func (s *ObservationSource) Kind() event.Kind { return event.Kind("observation") } func (s *ObservationSource) Poll(ctx context.Context) ([]event.Event, error) { // Re-check policy defensively (in case the URL is mutated after construction). if err := requireMetricUnits(s.http.URL); err != nil { return nil, fmt.Errorf("%s %q: %w", s.http.Driver, s.http.Name, err) } raw, meta, err := s.fetchRaw(ctx) if err != nil { return nil, err } eventID := buildEventID(s.http.Name, meta) if strings.TrimSpace(eventID) == "" { eventID = fmt.Sprintf("openweather:current:%s:%s", s.http.Name, time.Now().UTC().Format(time.RFC3339Nano)) } var effectiveAt *time.Time if !meta.ParsedTimestamp.IsZero() { t := meta.ParsedTimestamp.UTC() effectiveAt = &t } return common.SingleRawEvent( s.Kind(), s.http.Name, standards.SchemaRawOpenWeatherCurrentV1, eventID, effectiveAt, raw, ) } // ---- RAW fetch + minimal metadata decode ---- type openWeatherMeta struct { Dt int64 `json:"dt"` // unix seconds, UTC ID int64 `json:"id"` Name string `json:"name"` Coord struct { Lon float64 `json:"lon"` Lat float64 `json:"lat"` } `json:"coord"` ParsedTimestamp time.Time `json:"-"` } func (s *ObservationSource) fetchRaw(ctx context.Context) (json.RawMessage, openWeatherMeta, error) { raw, err := s.http.FetchJSON(ctx) if err != nil { return nil, openWeatherMeta{}, err } var meta openWeatherMeta if err := json.Unmarshal(raw, &meta); err != nil { return raw, openWeatherMeta{}, nil } if meta.Dt > 0 { meta.ParsedTimestamp = time.Unix(meta.Dt, 0).UTC() } return raw, meta, nil } func buildEventID(sourceName string, meta openWeatherMeta) string { locKey := "" if meta.ID != 0 { locKey = fmt.Sprintf("city:%d", meta.ID) } else if meta.Coord.Lat != 0 || meta.Coord.Lon != 0 { locKey = fmt.Sprintf("coord:%.5f,%.5f", meta.Coord.Lat, meta.Coord.Lon) } else { locKey = "loc:unknown" } ts := meta.ParsedTimestamp if ts.IsZero() { ts = time.Now().UTC() } return fmt.Sprintf("openweather:current:%s:%s:%s", sourceName, locKey, ts.Format(time.RFC3339Nano)) } func requireMetricUnits(rawURL string) error { u, err := url.Parse(strings.TrimSpace(rawURL)) if err != nil { return fmt.Errorf("invalid url %q: %w", rawURL, err) } units := strings.ToLower(strings.TrimSpace(u.Query().Get("units"))) if units != "metric" { if units == "" { units = "(missing; defaults to standard)" } return fmt.Errorf("url must include units=metric (got units=%s)", units) } return nil }