Rework interfaces and move lard.Timeseries to kdvh.LardTimeseries

migrations/kdvh/import.go

@@ -162,7 +162,7 @@ func (table *Table) importStation(station os.DirEntry, pool *pgxpool.Pool, confi
 				return
 			}
 
-			ts := lard.NewTimeseries(tsid, data)
+			ts := NewTimeseries(tsid, data)
 			count, err := importData(ts, tsInfo, pool, config)
 			if err != nil {
 				return
@@ -175,7 +175,7 @@ func (table *Table) importStation(station os.DirEntry, pool *pgxpool.Pool, confi
 	return totRows, nil
 }
 
-func (table *Table) parseElementFile(filename string, tsInfo *TimeseriesInfo, config *ImportConfig) ([]lard.Obs, error) {
+func (table *Table) parseElementFile(filename string, tsInfo *TimeseriesInfo, config *ImportConfig) ([]LardObs, error) {
 	file, err := os.Open(filename)
 	if err != nil {
 		slog.Warn(fmt.Sprintf("Could not open file '%s': %s", filename, err))
@@ -197,7 +197,7 @@ func (table *Table) parseElementFile(filename string, tsInfo *TimeseriesInfo, co
 	return data, nil
 }
 
-func importData(ts *lard.Timeseries, tsInfo *TimeseriesInfo, pool *pgxpool.Pool, config *ImportConfig) (count int64, err error) {
+func importData(ts *LardTimeseries, tsInfo *TimeseriesInfo, pool *pgxpool.Pool, config *ImportConfig) (count int64, err error) {
 	if !(config.Skip == "data") {
 		if tsInfo.param.IsScalar {
 			count, err = lard.InsertData(ts, pool, tsInfo.logstr)
@@ -206,7 +206,7 @@ func importData(ts *lard.Timeseries, tsInfo *TimeseriesInfo, pool *pgxpool.Pool,
 				return 0, err
 			}
 		} else {
-			count, err = lard.InsertNonscalarData(ts, pool, tsInfo.logstr)
+			count, err = lard.InsertTextData(ts, pool, tsInfo.logstr)
 			if err != nil {
 				slog.Error(tsInfo.logstr + "failed non-scalar data bulk insertion - " + err.Error())
 				return 0, err
@@ -217,13 +217,12 @@ func importData(ts *lard.Timeseries, tsInfo *TimeseriesInfo, pool *pgxpool.Pool,
 	}
 
 	if !(config.Skip == "flags") {
-		if err := lard.InsertFlags(ts, pool, tsInfo.logstr); err != nil {
+		if err := lard.InsertFlags(ts, FLAGS_TABLE, FLAGS_COLS, pool, tsInfo.logstr); err != nil {
 			slog.Error(tsInfo.logstr + "failed flag bulk insertion - " + err.Error())
 		}
 	}
 
 	return count, nil
-
 }
 
 func getStationNumber(station os.DirEntry, stationList []string) (int32, error) {
@@ -273,7 +272,7 @@ func getTimeseriesID(tsInfo *TimeseriesInfo, pool *pgxpool.Pool) (int32, error)
 	return tsid, nil
 }
 
-func (table *Table) parseData(handle *os.File, meta *TimeseriesInfo, config *ImportConfig) ([]lard.Obs, error) {
+func (table *Table) parseData(handle *os.File, meta *TimeseriesInfo, config *ImportConfig) ([]LardObs, error) {
 	scanner := bufio.NewScanner(handle)
 
 	var rowCount int
@@ -286,7 +285,7 @@ func (table *Table) parseData(handle *os.File, meta *TimeseriesInfo, config *Imp
 		}
 	}
 
-	data := make([]lard.Obs, 0, rowCount)
+	data := make([]LardObs, 0, rowCount)
 	for scanner.Scan() {
 		cols := strings.Split(scanner.Text(), config.Sep)
 

migrations/kdvh/import_functions.go

@@ -2,7 +2,6 @@ package kdvh
 
 import (
 	"errors"
-	"migrate/lard"
 	"strconv"
 
 	"github.com/rickb777/period"
@@ -251,18 +250,18 @@ func (obs *Obs) Useinfo() string {
 // and `useinfo` generated by Kvalobs for the observation, based on `Obs.Flags` and `Obs.Data`
 // Different KDVH tables need different ways to perform this conversion.
 
-func makeDataPage(obs Obs) (lard.Obs, error) {
+func makeDataPage(obs Obs) (LardObs, error) {
 	var valPtr *float32
 
 	controlinfo := VALUE_PASSED_QC
 	if obs.Data == "" {
 		controlinfo = VALUE_MISSING
 	}
 
-	// NOTE: this is the only function that can return `lard.Obs`
+	// NOTE: this is the only function that can return `LardObs`
 	// with non-null text data
 	if !obs.param.IsScalar {
-		return lard.Obs{
+		return LardObs{
 			Obstime:     obs.Obstime,
 			Data:        valPtr,
 			Text:        &obs.Data,
@@ -277,7 +276,7 @@ func makeDataPage(obs Obs) (lard.Obs, error) {
 		valPtr = &f32
 	}
 
-	return lard.Obs{
+	return LardObs{
 		Obstime:     obs.Obstime,
 		Data:        valPtr,
 		Useinfo:     obs.Useinfo(),
@@ -286,7 +285,7 @@ func makeDataPage(obs Obs) (lard.Obs, error) {
 }
 
 // modify obstimes to always use totime
-func makeDataPageProduct(obs Obs) (lard.Obs, error) {
+func makeDataPageProduct(obs Obs) (LardObs, error) {
 	obsLard, err := makeDataPage(obs)
 	if !obs.offset.IsZero() {
 		if temp, ok := obs.offset.AddTo(obsLard.Obstime); ok {
@@ -296,7 +295,7 @@ func makeDataPageProduct(obs Obs) (lard.Obs, error) {
 	return obsLard, err
 }
 
-func makeDataPageEdata(obs Obs) (lard.Obs, error) {
+func makeDataPageEdata(obs Obs) (LardObs, error) {
 	var controlinfo string
 	var valPtr *float32
 
@@ -314,15 +313,15 @@ func makeDataPageEdata(obs Obs) (lard.Obs, error) {
 		valPtr = &f32
 	}
 
-	return lard.Obs{
+	return LardObs{
 		Obstime:     obs.Obstime,
 		Data:        valPtr,
 		Useinfo:     obs.Useinfo(),
 		Controlinfo: controlinfo,
 	}, nil
 }
 
-func makeDataPagePdata(obs Obs) (lard.Obs, error) {
+func makeDataPagePdata(obs Obs) (LardObs, error) {
 	var controlinfo string
 	var valPtr *float32
 
@@ -355,15 +354,15 @@ func makeDataPagePdata(obs Obs) (lard.Obs, error) {
 
 	}
 
-	return lard.Obs{
+	return LardObs{
 		Obstime:     obs.Obstime,
 		Data:        valPtr,
 		Useinfo:     obs.Useinfo(),
 		Controlinfo: controlinfo,
 	}, nil
 }
 
-func makeDataPageNdata(obs Obs) (lard.Obs, error) {
+func makeDataPageNdata(obs Obs) (LardObs, error) {
 	var controlinfo string
 	var valPtr *float32
 
@@ -396,15 +395,15 @@ func makeDataPageNdata(obs Obs) (lard.Obs, error) {
 		valPtr = &f32
 	}
 
-	return lard.Obs{
+	return LardObs{
 		Obstime:     obs.Obstime,
 		Data:        valPtr,
 		Useinfo:     obs.Useinfo(),
 		Controlinfo: controlinfo,
 	}, nil
 }
 
-func makeDataPageVdata(obs Obs) (lard.Obs, error) {
+func makeDataPageVdata(obs Obs) (LardObs, error) {
 	var useinfo, controlinfo string
 	var valPtr *float32
 
@@ -426,11 +425,11 @@ func makeDataPageVdata(obs Obs) (lard.Obs, error) {
 			// add custom offset, because OT_24 in KDVH has been treated differently than OT_24 in kvalobs
 			offset, err := period.Parse("PT18H") // fromtime_offset -PT6H, timespan P1D
 			if err != nil {
-				return lard.Obs{}, errors.New("could not parse period")
+				return LardObs{}, errors.New("could not parse period")
 			}
 			temp, ok := offset.AddTo(obs.Obstime)
 			if !ok {
-				return lard.Obs{}, errors.New("could not add period")
+				return LardObs{}, errors.New("could not add period")
 			}
 
 			obs.Obstime = temp
@@ -441,22 +440,22 @@ func makeDataPageVdata(obs Obs) (lard.Obs, error) {
 		controlinfo = VALUE_PASSED_QC
 	}
 
-	return lard.Obs{
+	return LardObs{
 		Obstime:     obs.Obstime,
 		Data:        valPtr,
 		Useinfo:     useinfo,
 		Controlinfo: controlinfo,
 	}, nil
 }
 
-func makeDataPageDiurnalInterpolated(obs Obs) (lard.Obs, error) {
+func makeDataPageDiurnalInterpolated(obs Obs) (LardObs, error) {
 	val, err := strconv.ParseFloat(obs.Data, 32)
 	if err != nil {
-		return lard.Obs{}, err
+		return LardObs{}, err
 	}
 	f32 := float32(val)
 
-	return lard.Obs{
+	return LardObs{
 		Obstime:     obs.Obstime,
 		Data:        &f32,
 		Useinfo:     DIURNAL_INTERPOLATED_USEINFO,

migrations/kdvh/lard.go

@@ -0,0 +1,63 @@
+package kdvh
+
+import (
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// LardTimeseries in LARD have and ID and associated observations
+type LardTimeseries struct {
+	id   int32
+	data []LardObs
+}
+
+func NewTimeseries(id int32, data []LardObs) *LardTimeseries {
+	return &LardTimeseries{id, data}
+}
+
+func (ts *LardTimeseries) Len() int {
+	return len(ts.data)
+}
+
+func (ts *LardTimeseries) InsertData(i int) ([]any, error) {
+	return []any{
+		ts.id,
+		ts.data[i].Obstime,
+		ts.data[i].Data,
+	}, nil
+}
+
+func (ts *LardTimeseries) InsertText(i int) ([]any, error) {
+	return []any{
+		ts.id,
+		ts.data[i].Obstime,
+		ts.data[i].Text,
+	}, nil
+}
+
+var FLAGS_TABLE pgx.Identifier = pgx.Identifier{"flags", "kdvh"}
+var FLAGS_COLS []string = []string{"timeseries", "obstime", "controlinfo", "useinfo"}
+
+func (ts *LardTimeseries) InsertFlags(i int) ([]any, error) {
+	return []any{
+		ts.id,
+		ts.data[i].Obstime,
+		ts.data[i].Controlinfo,
+		ts.data[i].Useinfo,
+	}, nil
+}
+
+// Struct containg all the fields we want to save in LARD from KDVH
+type LardObs struct {
+	// Time of observation
+	Obstime time.Time
+	// Observation data formatted as a single precision floating point number
+	Data *float32
+	// Observation data that cannot be represented as a float, therefore stored as a string
+	Text *string
+	// Flag encoding quality control status
+	Controlinfo string
+	// Flag encoding quality control status
+	Useinfo string
+}

migrations/kdvh/table.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
-	"migrate/lard"
 	"time"
 
 	"github.com/rickb777/period"
@@ -35,7 +34,7 @@ type Table struct {
 	ElemTableName string          // Name of the ELEM table
 	Path          string          // Directory name of where the dumped table is stored
 	dumpFunc      DumpFunction    // Function used to dump the KDVH table (found in `dump_functions.go`)
-	convFunc      ConvertFunction // Function that converts KDVH obs to LARD obs (found in `import_functions.go`)
+	convFunc      ConvertFunction // Function that converts KDVH obs to Lardobs (found in `import_functions.go`)
 	importUntil   int             // Import data only until the year specified by this field
 }
 
@@ -47,7 +46,7 @@ type DumpMeta struct {
 	flagTable string
 }
 
-type ConvertFunction func(Obs) (lard.Obs, error)
+type ConvertFunction func(Obs) (LardObs, error)
 type Obs struct {
 	*TimeseriesInfo
 	Obstime time.Time
@@ -118,7 +117,7 @@ func (t *Table) SetDumpFunc(fn DumpFunction) *Table {
 	return t
 }
 
-// Sets the function used to convert observations from the table to LARD observations
+// Sets the function used to convert observations from the table to Lardobservations
 func (t *Table) SetConvFunc(fn ConvertFunction) *Table {
 	t.convFunc = fn
 	return t