Port functions from VSTP to handle CIF data. Added tests for Cif/helpers functions

src/cif/helpers.go

@@ -68,3 +68,36 @@ func generateUpdateDays(days int) []time.Time {
 
 	return updateDays
 }
+
+// Parses CIF Schedule Start/End Dates (YYYY-MM-DD) into time.Time types (00:00:00 for start, 23:59:59 for end)
+func ParseCifDate(input, startOrEnd string) time.Time {
+	layout := "2006-01-02" // Layout of input
+	t, err := time.ParseInLocation(layout, input, londonTimezone)
+	if err != nil {
+		log.Msg.Error("Error parsing date string", zap.String("date string", input), zap.Error(err))
+		return time.Time{}
+	}
+
+	if startOrEnd == "start" {
+		t = time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, londonTimezone)
+	} else if startOrEnd == "end" {
+		t = time.Date(t.Year(), t.Month(), t.Day(), 23, 59, 59, 0, londonTimezone)
+	} else {
+		log.Msg.Error("Error parsing date string", zap.String("date string", input), zap.Error(err))
+		return time.Time{}
+	}
+
+	return t
+}
+
+// Parses CIF days_run field and converts to array of day strings
+func parseDaysRun(daysBinary string) []string {
+	shortDays := []string{"m", "t", "w", "th", "f", "s", "su"}
+	var result []string
+	for i, digit := range daysBinary {
+		if digit == '1' {
+			result = append(result, shortDays[i])
+		}
+	}
+	return result
+}

src/cif/helpers_test.go

@@ -1,6 +1,7 @@
 package cif
 
 import (
+	"reflect"
 	"testing"
 	"time"
 )
@@ -86,6 +87,48 @@ func TestGenerateUpdateDays(t *testing.T) {
 	}
 }
 
+func TestParseCifDate(t *testing.T) {
+	testCases := []struct {
+		dateString string
+		startOrEnd string
+		expect     time.Time
+	}{
+		{"2024-04-05", "start", time.Date(2024, time.April, 5, 0, 0, 0, 0, londonTimezone)},
+		{"2022-01-01", "start", time.Date(2022, time.January, 1, 0, 0, 0, 0, londonTimezone)},
+		{"2015-09-26", "end", time.Date(2015, time.September, 26, 23, 59, 59, 0, londonTimezone)},
+		{"2018-03-13", "end", time.Date(2018, time.March, 13, 23, 59, 59, 0, londonTimezone)},
+	}
+
+	layout := "2006-01-02 15:04:05" // Layout for printing times in error cases.
+
+	for _, tc := range testCases {
+		result := ParseCifDate(tc.dateString, tc.startOrEnd)
+		if result != tc.expect {
+			t.Errorf("For datestring %s, startOrEnd %s, expected %s, but got %s", tc.dateString, tc.startOrEnd, tc.expect.Format(layout), result.Format(layout))
+		}
+	}
+}
+
+func TestParseDaysRun(t *testing.T) {
+	testCases := []struct {
+		input  string
+		expect []string
+	}{
+		{"1111111", []string{"m", "t", "w", "th", "f", "s", "su"}},
+		{"0000001", []string{"su"}},
+		{"1000000", []string{"m"}},
+		{"0000100", []string{"f"}},
+		{"0111000", []string{"t", "w", "th"}},
+	}
+
+	for _, tc := range testCases {
+		result := parseDaysRun(tc.input)
+		if !reflect.DeepEqual(result, tc.expect) {
+			t.Errorf("For input %s, expected %v, but got %v", tc.input, tc.expect, result)
+		}
+	}
+}
+
 // Checks if two time values have the same year, month and day.
 func isSameDate(t1, t2 time.Time) bool {
 	return t1.Year() == t2.Year() && t1.Month() == t2.Month() && t1.Day() == t2.Day()

src/cif/parse_test.go

@@ -1,105 +0,0 @@
-package cif
-
-import (
-	"testing"
-	"time"
-
-	"git.fjla.uk/owlboard/go-types/pkg/upstreamApi"
-)
-
-func TestParseCifData(t *testing.T) {
-	testCases := []struct {
-		name     string
-		input    []byte
-		expected *parsedData
-	}
-}
-
-func returnParsedData() *parsedData {
-	var parsed = {
-		name: "Valid JSON Data 1",
-		input: []byte(`
-		    {"JsonTimetableV1":{"classification":"public","timestamp":1711227636,"owner":"Network Rail","Sender":{"organisation":"Rockshore","application":"NTROD","component":"SCHEDULE"},"Metadata":{"type":"update","sequence":4307}}}
-			{"JsonAssociationV1":{"transaction_type":"Create","main_train_uid":"L65283","assoc_train_uid":"L65245","assoc_start_date":"2023-12-11T00:00:00Z","assoc_end_date":"2024-03-28T00:00:00Z","assoc_days":"1111000","category":"  ","date_indicator":" ","location":"RDNGSTN","base_location_suffix":null,"assoc_location_suffix":null,"diagram_type":"T","CIF_stp_indicator":"C"}}
-            {"JsonAssociationV1":{"transaction_type":"Delete","main_train_uid":"L65283","assoc_train_uid":"L65245","assoc_start_date":"2024-03-18T00:00:00Z","location":"RDNGSTN","base_location_suffix":null,"diagram_type":"T","CIF_stp_indicator":"C"}}
-			{"JsonScheduleV1":{"CIF_stp_indicator":"O","CIF_train_uid":"C25513","schedule_start_date":"2024-03-23","transaction_type":"Delete"}}
-			{"JsonScheduleV1":{"CIF_bank_holiday_running":null,"CIF_stp_indicator":"O","CIF_train_uid":"Y04345","applicable_timetable":"Y","atoc_code":"SW","new_schedule_segment":{"traction_class":"","uic_code":""},"schedule_days_runs":"0000001","schedule_end_date":"2024-05-19","schedule_segment":{"signalling_id":"5A20","CIF_train_category":"EE","CIF_headcode":"00","CIF_course_indicator":1,"CIF_train_service_code":"24676004","CIF_business_sector":"??","CIF_power_type":"EMU","CIF_timing_load":null,"CIF_speed":"100","CIF_operating_characteristics":"D","CIF_train_class":null,"CIF_sleepers":null,"CIF_reservations":null,"CIF_connection_indicator":null,"CIF_catering_code":null,"CIF_service_branding":"","schedule_location":[{"location_type":"LO","record_identity":"LO","tiploc_code":"FARNHMD","tiploc_instance":null,"departure":"0721","public_departure":null,"platform":null,"line":null,"engineering_allowance":null,"pathing_allowance":null,"performance_allowance":null},{"location_type":"LT","record_identity":"LT","tiploc_code":"FARNHAM","tiploc_instance":null,"platform":"1","arrival":"0729","public_arrival":null,"path":null}]},"schedule_start_date":"2024-05-19","train_status":"T","transaction_type":"Create"}}		
-			{"EOF":true}
-		`),
-		expected: &parsedData{
-			header: upstreamApi.JsonTimetableV1{
-				Classification: "public",
-				Timestamp:      1711227636,
-				Owner:          "Network Rail",
-				Sender: upstreamApi.TimetableSender{
-					Organisation: "Rockshore",
-					Application:  "NTROD",
-					Component:    "SCHEDULE",
-				},
-				Metadata: upstreamApi.TimetableMetadata{
-					Type:     "update",
-					Sequence: 4307,
-				},
-			},
-			assoc: []upstreamApi.JsonAssociationV1{
-				TransactionType: "Create",
-				MainTrainUid:    "L65283",
-				AssocTrainUid:   "L65245",
-				AssocStartDate:  time.Date(2023, time.December, 11, 0, 0, 0, 0, time.UTC),
-				AssocEndDate:    time.Date(2024, time.March, 28, 0, 0, 0, 0, time.UTC),
-				AssocDays:       "1111000",
-				Location:        "RDNGSTN",
-				DiagramType:     "T",
-				CifStpIndicator: "C",
-			}, {
-				TransactionType: "Delete",
-				MainTrainUid:    "L65283",
-				AssocTrainUid:   "L65245",
-				AssocStartDate:  time.Date(2024, time.March, 18, 0, 0, 0, 0, time.UTC),
-				Location:        "RDNGSTN",
-				DiagramType:     "T",
-				CifStpIndicator: "C",
-			},
-			sched: []upstreamApi.JsonScheduleV1{
-				CifStpIndicator:   "O",
-				CifTrainUid:       "C25513",
-				ScheduleStartDate: "2024-03-23",
-				TransactionType:   "Delete",
-			}, {
-				CifStpIndicator:     "O",
-				CifTrainUid:         "Y04345",
-				ApplicableTimetable: "Y",
-				AtocCode:            "SW",
-				NewScheduleSegment:  upstreamApi.CifNewScheduleSegment{},
-				ScheduleDaysRun:     "0000001",
-				ScheduleEndDate:     "2024-05-19",
-				ScheduleSegment: upstreamApi.CifScheduleSegment{
-					SignallingId:                "5A20",
-					CifTrainCategory:            "EE",
-					CifHeadcode:                 "OO",
-					CifCourseIndicator:          1,
-					CifTrainServiceCode:         "24676004",
-					CifBusinessSector:           "??",
-					CifPowerType:                "EMU",
-					CifSpeed:                    "100",
-					CifOperatingCharacteristics: "D",
-					ScheduleLocation: []upstreamApi.CifScheduleLocation{
-						LocationType:   "LO",
-						RecordIdentity: "LO",
-						TiplocCode:     "FARNHMD",
-						Departure:      "0721",
-					}, {
-						LocationType:   "LT",
-						RecordIdentity: "LT",
-						TiplocCode:     "FARNHAM",
-						Platform:       "1",
-						Arrival:        "0729",
-					},
-				},
-				ScheduleStartDate: "2024-05-19",
-				TrainStatus:       "T",
-				TransactionType:   "Create",
-			},
-		},
-	}
-}

src/cif/process.go

@@ -1,6 +1,68 @@
 package cif
 
+import (
+	"git.fjla.uk/owlboard/go-types/pkg/database"
+	"git.fjla.uk/owlboard/go-types/pkg/upstreamApi"
+	"git.fjla.uk/owlboard/timetable-mgr/log"
+	"go.uber.org/zap"
+)
+
 // Processes parsed CIF data and applies the data to the database
 func processParsedCif(data *parsedData) error {
+	createTasks := make([]*upstreamApi.JsonScheduleV1, 0)
+	deleteTasks := make([]*upstreamApi.JsonScheduleV1, 0)
+
+	for _, item := range data.sched {
+		switch item.TransactionType {
+		case "Delete":
+			deleteTasks = append(deleteTasks, &item)
+		case "Create":
+			createTasks = append(createTasks, &item)
+		default:
+			log.Msg.Error("Unknown transaction type in CIF Schedule", zap.String("TransactionType", item.TransactionType))
+		}
+	}
+
+	err := doDeletions(deleteTasks)
+	if err != nil {
+		log.Msg.Error("Error deleting CIF Entries", zap.Error(err))
+		return err
+	}
+	err = doCreations(createTasks)
+	if err != nil {
+		log.Msg.Error("Error creating CIF Entries", zap.Error(err))
+		return err
+	}
+
+	return nil
+}
+
+// Create delete query types and pass to the function which batches the deletions
+func doDeletions(deletions []*upstreamApi.JsonScheduleV1) error {
+	log.Msg.Info("Preparing CIF update Delete tasks", zap.Int("Delete task count", len(deletions)))
+
+	deleteQueries := make([]database.DeleteQuery, 0)
+	for _, item := range deletions {
+		query := database.DeleteQuery{
+			ScheduleStartDate: ParseCifDate(item.ScheduleStartDate, "start"),
+			StpIndicator:      item.CifStpIndicator,
+			TrainUid:          item.CifTrainUid,
+		}
+
+		deleteQueries = append(deleteQueries, query)
+	}
+	return nil
+}
+
+// Convert to the correct struct for the database and pass to the function which batches insertions
+func doCreations(creations []*upstreamApi.JsonScheduleV1) error {
+	log.Msg.Info("Preparing CIF update Create tasks", zap.Int("Create task count", len(creations)))
+
+	createDocuments := make([]database.Service, 0)
+	for _, item := range creations {
+		document := database.Service{}
+
+		createDocuments = append(createDocuments, document)
+	}
 	return nil
 }

src/cif/update.go

@@ -60,7 +60,7 @@ func runCifUpdateDownload(cfg *helpers.Configuration, metadata *dbAccess.CifMeta
 			return err
 		}
 		// Parse CIF file
-		parsed, err := parseCifData(data)
+		parsed, err = parseCifData(data)
 		if err != nil {
 			log.Msg.Error("Error parsing CIF data", zap.Error(err))
 			return err

src/dbAccess/cif.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"time"
 
+	"git.fjla.uk/owlboard/go-types/pkg/database"
 	"git.fjla.uk/owlboard/timetable-mgr/log"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
@@ -62,3 +63,31 @@ func PutCifMetadata(metadata CifMetadata) bool {
 	}
 	return true
 }
+
+func DeleteCifEntries(deletions []database.DeleteQuery) error {
+	// Prepare deletion tasks
+	bulkDeletions := make([]mongo.WriteModel, 0, len(deletions))
+
+	for _, deleteQuery := range deletions {
+		filter := bson.M{
+			"trainUid":          deleteQuery.TrainUid,
+			"scheduleStartDate": deleteQuery.ScheduleStartDate,
+			"stpIndicator":      deleteQuery.StpIndicator,
+		}
+		bulkDeletions = append(bulkDeletions, mongo.NewDeleteManyModel().SetFilter(filter))
+	}
+
+	log.Msg.Info("Running `Delete` tasks from CIF Update", zap.Int("Required deletions", len(deletions)))
+	for i := 0; i < len(bulkDeletions); i += batchsize {
+		end := i + batchsize
+		if end > len(bulkDeletions) {
+			end = len(bulkDeletions)
+		}
+		_, err := MongoClient.Database(databaseName).Collection(TimetableCollection).BulkWrite(context.Background(), bulkDeletions[i:end])
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

src/dbAccess/contants.go

@@ -5,3 +5,4 @@ const CorpusCollection string = "corpus"
 const StationsCollection string = "stations"
 const metaCollection string = "meta"
 const TimetableCollection string = "timetable"
+const batchsize int = 100

src/go.mod

@@ -3,7 +3,7 @@ module git.fjla.uk/owlboard/timetable-mgr
 go 1.21
 
 require (
-	git.fjla.uk/owlboard/go-types v0.0.0-20240403200521-41796e25b6c3
+	git.fjla.uk/owlboard/go-types v0.0.0-20240405194933-7dafca950460
 	github.com/go-stomp/stomp/v3 v3.0.5
 	go.mongodb.org/mongo-driver v1.12.0
 	go.uber.org/zap v1.24.0

src/go.sum

@@ -4,6 +4,8 @@ git.fjla.uk/owlboard/go-types v0.0.0-20240331204922-8f8899eb6072 h1:QjaTVm4bpnXZ
 git.fjla.uk/owlboard/go-types v0.0.0-20240331204922-8f8899eb6072/go.mod h1:kG+BX9UF+yJaAVnln/QSKlTdrtKRRReezMeSk1ZLMzY=
 git.fjla.uk/owlboard/go-types v0.0.0-20240403200521-41796e25b6c3 h1:veGmL8GeWsgGCeTgPPSDw5tbUr1pUz8F6DBgFMf6IGc=
 git.fjla.uk/owlboard/go-types v0.0.0-20240403200521-41796e25b6c3/go.mod h1:kG+BX9UF+yJaAVnln/QSKlTdrtKRRReezMeSk1ZLMzY=
+git.fjla.uk/owlboard/go-types v0.0.0-20240405194933-7dafca950460 h1:39vcejk0MzZIL2WdriKSfcDoCc/0NWzZMncHqw5ctbY=
+git.fjla.uk/owlboard/go-types v0.0.0-20240405194933-7dafca950460/go.mod h1:kG+BX9UF+yJaAVnln/QSKlTdrtKRRReezMeSk1ZLMzY=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

src/vstp/parser.go

@@ -101,7 +101,7 @@ func parseDate(dateString string, end bool) time.Time {
 	var hour, minute, second, nanosecond int
 	location := time.UTC
 	if end {
-		hour, minute, second, nanosecond = 23, 59, 59, 999999999
+		hour, minute, second, nanosecond = 23, 59, 59, 0
 	} else {
 		hour, minute, second, nanosecond = 0, 0, 0, 0
 	}