1 files changed, 82 insertions, 0 deletions
diff --git a/internal/weather/utils.go b/internal/weather/utils.go
new file mode 100644
index 0000000..b59949a
--- /dev/null
+++ b/internal/weather/utils.go
@@ -0,0 +1,82 @@
+// internal/weather/utils.go
+package weather
+
+import (
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// GetEnv gets environment variable with default
+func GetEnv(key, defaultValue string) string {
+	if value := os.Getenv(key); value != "" {
+		return value
+	}
+	return defaultValue
+}
+
+// ParseDuration parses duration string
+func ParseDuration(s string) time.Duration {
+	d, err := time.ParseDuration(s)
+	if err != nil {
+		return 5 * time.Minute
+	}
+	return d
+}
+
+// ParseInt parses integer string
+func ParseInt(s string) int {
+	i, err := strconv.Atoi(s)
+	if err != nil {
+		return 0
+	}
+	return i
+}
+
+// ParseFloat parses float string
+func ParseFloat(s string) float64 {
+	f, err := strconv.ParseFloat(s, 64)
+	if err != nil {
+		return 0
+	}
+	return f
+}
+
+// ParseTopics parses comma-separated topics
+func ParseTopics(s string) []string {
+	if s == "" {
+		return []string{
+			"weather/obs/fmi/#",
+			"weather/forecast/fmi/#",
+		}
+	}
+	return strings.Split(s, ",")
+}
+
+// LoadWatermark loads last observation time from file
+func LoadWatermark(filename string) time.Time {
+	b, err := os.ReadFile(filename)
+	if err != nil {
+		return time.Now().UTC().Add(-24 * time.Hour)
+	}
+	sec, err := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64)
+	if err != nil {
+		return time.Now().UTC().Add(-24 * time.Hour)
+	}
+	return time.Unix(sec, 0).UTC()
+}
+
+// StoreWatermark stores observation time to file
+func StoreWatermark(filename string, t time.Time) error {
+	data := []byte(strconv.FormatInt(t.Unix(), 10))
+	return os.WriteFile(filename, data, 0644)
+}
+
+// InitTopicStats initializes topic statistics
+func InitTopicStats() *TopicStats {
+	return &TopicStats{
+		MessagesReceived: make(map[string]int),
+		LastMessageTime:  make(map[string]time.Time),
+	}
+}