Add meta stats

studies/db/go_rewrite/main.go

@@ -1,11 +1,12 @@
 package main
 
 import (
-	"studies/runner"
+	"studies/times"
 )
 
 func main() {
-	runner.Run()
+	//runner.Run()
+	times.DebugPrint()
 	//times.DebugPrint()
 	// Wait forever.
 	// select {}

studies/db/go_rewrite/notebooks/studies_meta.ipynb

@@ -0,0 +1,90 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 4,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Collecting psycopg2-binary\n",
+      "  Downloading psycopg2_binary-2.9.9-cp311-cp311-macosx_11_0_arm64.whl (2.6 MB)\n",
+      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m2.6/2.6 MB\u001b[0m \u001b[31m5.9 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0ma \u001b[36m0:00:01\u001b[0m\n",
+      "\u001b[?25hInstalling collected packages: psycopg2-binary\n",
+      "Successfully installed psycopg2-binary-2.9.9\n",
+      "\n",
+      "\u001b[1m[\u001b[0m\u001b[34;49mnotice\u001b[0m\u001b[1;39;49m]\u001b[0m\u001b[39;49m A new release of pip is available: \u001b[0m\u001b[31;49m23.0.1\u001b[0m\u001b[39;49m -> \u001b[0m\u001b[32;49m23.3.1\u001b[0m\n",
+      "\u001b[1m[\u001b[0m\u001b[34;49mnotice\u001b[0m\u001b[1;39;49m]\u001b[0m\u001b[39;49m To update, run: \u001b[0m\u001b[32;49mpython3.11 -m pip install --upgrade pip\u001b[0m\n"
+     ]
+    }
+   ],
+   "source": [
+    "!pip3 install psycopg2-binary"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 5,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Rows in 'observation_dbs' table\n",
+      "(3350124833,)\n"
+     ]
+    }
+   ],
+   "source": [
+    "import psycopg2\n",
+    "\n",
+    "conn = psycopg2.connect(\n",
+    "    host=\"priobike-sentry.inf.tu-dresden.de\",\n",
+    "    port=443,\n",
+    "    database=\"observations\",\n",
+    "    user=\"postgres\",\n",
+    "    password=\"Et7RvZ4TjEBHRF\")\n",
+    "\n",
+    "cur = conn.cursor()\n",
+    "cur.execute('SELECT COUNT(*) FROM observation_dbs;')\n",
+    "rows_observation_dbs = cur.fetchone()\n",
+    "print(\"Rows in 'observation_dbs' table\")\n",
+    "print(rows_observation_dbs)\n",
+    "\n",
+    "cur.close()\n",
+    "conn.close()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3 (ipykernel)",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.11.4"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

studies/db/go_rewrite/runner/runner_complete_cell.go

@@ -151,6 +151,7 @@ func RunCompleteCell(tldThings []things.TLDThing, suffixName string) map[string]
 				processedThings[name].MedianShifts[dayIdx] = thing.MedianShifts[dayIdx]
 				processedThings[name].MetricsRelativeGreenDistance[dayIdx] = thing.MetricsRelativeGreenDistance[dayIdx]
 				processedThings[name].MedianGreenLengths[dayIdx] = thing.MedianGreenLengths[dayIdx]
+				processedThings[name].Results[dayIdx] = thing.Results[dayIdx]
 			}
 		}
 	}

studies/db/go_rewrite/runner/runner_meta_stats.go

@@ -0,0 +1,188 @@
+package runner
+
+import (
+	// "time"
+	"encoding/json"
+	"os"
+	"sync"
+
+	"studies/db"
+	"studies/things"
+	"studies/times"
+)
+
+func RunMetaStats(tldThings []things.TLDThing) (uint64, uint64, []string) {
+	statsByRoutines := [7]map[string]uint64{}
+	thingsWithObservations := [7][]string{}
+
+	for idx := range statsByRoutines {
+		statsByRoutines[idx] = map[string]uint64{
+			"ps_observation_count": 0,
+			"cs_observation_count": 0,
+		}
+		thingsWithObservations[idx] = []string{}
+	}
+
+	datastreamsByThingName := map[string]map[string]int32{}
+	thingsByDatastreamId := map[int32]string{}
+	for _, tldThing := range tldThings {
+		for i := 0; i < len(tldThing.Datastreams); i++ {
+			if tldThing.Datastreams[i].Properties.LayerName == "primary_signal" || tldThing.Datastreams[i].Properties.LayerName == "secondary_signal" {
+				if _, ok := datastreamsByThingName[tldThing.Name]; !ok {
+					datastreamsByThingName[tldThing.Name] = map[string]int32{}
+				}
+				datastreamsByThingName[tldThing.Name]["primary_signal"] = tldThing.Datastreams[i].ID
+				thingsByDatastreamId[tldThing.Datastreams[i].ID] = tldThing.Name
+			} else if tldThing.Datastreams[i].Properties.LayerName == "cycle_second" {
+				if _, ok := datastreamsByThingName[tldThing.Name]; !ok {
+					datastreamsByThingName[tldThing.Name] = map[string]int32{}
+				}
+				datastreamsByThingName[tldThing.Name]["cycle_second"] = tldThing.Datastreams[i].ID
+				thingsByDatastreamId[tldThing.Datastreams[i].ID] = tldThing.Name
+			} else {
+				panic("Unknown layer name")
+			}
+		}
+	}
+
+	pool := db.NewPool()
+
+	var wg sync.WaitGroup
+	// startTime := time.Now()
+	times := times.GetCells()
+	for i := 0; i < len(times); i++ {
+		client := pool.GetClient()
+		wg.Add(1)
+		go func(stats *map[string]uint64, thingsWithObservations *[]string, dayIdx int, day [24][4][2]int32, dbClient *db.Client) {
+			defer wg.Done()
+			for hourIdx, hour := range day {
+				println("Processing day", dayIdx, "hour", hourIdx)
+				cells := [4][2]int32{}
+				for cellIdx, cell := range hour {
+					// println(cell[0], " ", cell[1])
+					cells[cellIdx] = [2]int32{cell[0], cell[1]}
+				}
+				query := db.GetCellsAllDatastreamsQuery(cells)
+				rows := dbClient.Query(query)
+				// println("Receiving observations..")
+				currentCellIdx := 0
+				observationCount := 0
+				// rowCount := 0
+				for rows.Next() {
+					var phenomenon_time int32
+					var result int16
+					var datastream_id int32
+					err := rows.Scan(&phenomenon_time, &result, &datastream_id)
+					if err != nil {
+						panic(err)
+					}
+					if phenomenon_time > cells[currentCellIdx][1] {
+						currentCellIdx++
+
+						// println("Calc Cycles")
+					}
+					thingName := thingsByDatastreamId[datastream_id]
+					if datastream_id == datastreamsByThingName[thingName]["primary_signal"] {
+						(*stats)["ps_observation_count"]++
+					} else if datastream_id == datastreamsByThingName[thingName]["cycle_second"] {
+						(*stats)["cs_observation_count"]++
+					} else {
+						continue
+					}
+					observationCount++
+					/* rowCount++
+					if rowCount % 100000 == 0 {
+						println("Processed observations: ", rowCount)
+					} */
+				}
+				println("Observation count: ", observationCount)
+				rows.Close()
+				// println("Processed observations")
+				// endTime := time.Now()
+				// elapsed := endTime.Sub(startTime)
+				// println("Elapsed time in seconds: ", elapsed.Seconds())
+			}
+			dbClient.Close()
+		}(&statsByRoutines[i], &thingsWithObservations[i], i, times[i], client)
+	}
+
+	wg.Wait()
+
+	pool.Close()
+
+	stats := map[string]uint64{}
+	for _, statsByRoutines := range statsByRoutines {
+		if _, ok := stats["ps_observation_count"]; !ok {
+			stats["ps_observation_count"] = statsByRoutines["ps_observation_count"]
+		} else {
+			stats["ps_observation_count"] += statsByRoutines["ps_observation_count"]
+		}
+		if _, ok := stats["cs_observation_count"]; !ok {
+			stats["cs_observation_count"] = statsByRoutines["cs_observation_count"]
+		} else {
+			stats["cs_observation_count"] += statsByRoutines["cs_observation_count"]
+		}
+	}
+
+	things := []string{}
+	for _, thingsWithObservations := range thingsWithObservations {
+		for _, thing := range thingsWithObservations {
+			if !stringInSlice(things, thing) {
+				things = append(things, thing)
+			}
+		}
+	}
+
+	return stats["ps_observation_count"], stats["cs_observation_count"], things
+}
+
+func stringInSlice(s []string, e string) bool {
+	for _, a := range s {
+		if a == e {
+			return true
+		}
+	}
+	return false
+}
+
+func RunMeta() {
+	tp := things.NewThingsProvider(false)
+	tp.FilterOnlyPrimarySignalAndCycleSecondDatastreams()
+	tldThingsPrimary := tp.Things
+
+	psObservationCountTotal := uint64(0)
+	csObservationCountTotal := uint64(0)
+	thingsWithObservationsTotal := []string{}
+
+	psObservationCount, csObservationCount, thingsWithObservations := RunMetaStats(tldThingsPrimary)
+
+	thingsWithObservationsTotal = append(thingsWithObservationsTotal, thingsWithObservations...)
+
+	psObservationCountTotal += psObservationCount
+	csObservationCountTotal += csObservationCount
+
+	tp = things.NewThingsProvider(false)
+	tp.FilterOnlySecondarySignalAndCycleSecondDatastreams()
+	tldThingsSecondary := tp.Things
+
+	psObservationCount, csObservationCount, thingsWithObservations = RunMetaStats(tldThingsSecondary)
+
+	for _, thing := range thingsWithObservations {
+		if !stringInSlice(thingsWithObservationsTotal, thing) {
+			thingsWithObservationsTotal = append(thingsWithObservationsTotal, thing)
+		}
+	}
+
+	psObservationCountTotal += psObservationCount
+	csObservationCountTotal += csObservationCount
+
+	meta_stats := map[string]uint64{
+		"ps_observation_count_total": psObservationCountTotal,
+		"cs_observation_count_total": csObservationCountTotal,
+		"things_with_observations":   uint64(len(thingsWithObservationsTotal)),
+	}
+
+	// Output processed things as json file
+	file, _ := json.MarshalIndent(meta_stats, "", " ")
+	_ = os.WriteFile("meta_stats.json", file, 0644)
+}

studies/db/go_rewrite/things/thing.go

@@ -65,6 +65,7 @@ type Thing struct {
 	MetricsSP                    [7][24]float64
 	MedianShifts                 [7][24]float64
 	MedianGreenLengths           [7][24]float64
+	Results                      [7][24][]int8
 }
 
 func NewThing(name string, validation bool, retrieveAllCycleCleanupStats bool) *Thing {
@@ -96,6 +97,7 @@ func NewThing(name string, validation bool, retrieveAllCycleCleanupStats bool) *
 	thing.MetricsSP = [7][24]float64{}
 	thing.MedianShifts = [7][24]float64{}
 	thing.MedianGreenLengths = [7][24]float64{}
+	thing.Results = [7][24][]int8{}
 	return thing
 }
 
@@ -347,14 +349,29 @@ func (thing *Thing) getGreenLength(cycle cycle) float64 {
 	return greenLength
 }
 
+func resultInSlice(s []int8, e int8) bool {
+	for _, a := range s {
+		if a == e {
+			return true
+		}
+	}
+	return false
+}
+
 func (thing *Thing) CalculateMetrics(day int, hour int) {
 	distances := make([]float64, 0)
 	relativeGreenDistances := make([]float64, 0)
 	totalGreenDiffs := make([]float64, 0)
 	greenLengths := make([]float64, 0)
 	cycles := []cycle{}
+	results := make([]int8, 0)
 	for _, cellCycles := range thing.cycles {
 		for idx, cycle := range cellCycles {
+			for _, result := range cycle.results {
+				if !resultInSlice(results, result) {
+					results = append(results, result)
+				}
+			}
 			cycles = append(cycles, cycle)
 			greenIndices := thing.getGreenIndices(cycle)
 			greenLengths = append(greenLengths, thing.getGreenLength(cycle))
@@ -386,6 +403,8 @@ func (thing *Thing) CalculateMetrics(day int, hour int) {
 		}
 	}
 
+	thing.Results[day][hour] = results
+
 	if len(greenLengths) == 0 {
 		thing.MedianGreenLengths[day][hour] = -1.0
 	} else {