There should be no need to await.

That is because the queue handler will, in theory, only ever
process one event at a time.

bin/daily_tasks.py

@@ -11,11 +11,9 @@ from datastore import Datastore
 
 if __name__ == '__main__':
 
-	print("Reading configuration")
-	surveys = configuration.surveys()
-
 	print("Connecting to database")
 	db = Datastore()
+	surveys = db.surveys()
 
 	print("Reading surveys")
 	for survey in surveys:

lib/www/client/source/src/components/event-edit.vue

@@ -44,7 +44,7 @@
               <template v-slot:activator="{ on, attrs }">
                 <v-text-field
                   v-model="tsDate"
-                  :disabled="!!(sequence || point || entrySequence || entryPoint)"
+                  :disabled="!!(entrySequence || entryPoint)"
                   label="Date"
                   suffix="UTC"
                   prepend-icon="mdi-calendar"
@@ -64,7 +64,7 @@
           <v-col>
             <v-text-field
               v-model="tsTime"
-              :disabled="!!(sequence || point || entrySequence || entryPoint)"
+              :disabled="!!(entrySequence || entryPoint)"
               label="Time"
               suffix="UTC"
               prepend-icon="mdi-clock-outline"
@@ -256,6 +256,15 @@
         >
           Cancel
         </v-btn>
+        <v-btn v-if="!id && (entrySequence || entryPoint)"
+          color="info"
+          text
+          title="Enter an event by time"
+          @click="timed"
+        >
+        <v-icon left small>mdi-clock-outline</v-icon>
+        Timed
+        </v-btn>
         <v-spacer></v-spacer>
         <v-btn
           :disabled="!canSave"
@@ -632,6 +641,14 @@ export default {
       }
     },
 
+    timed () {
+      const tstamp = (new Date()).toISOString();
+      this.entrySequence = null;
+      this.entryPoint = null;
+      this.tsDate = tstamp.substr(0, 10);
+      this.tsTime = tstamp.substr(11, 8);
+    },
+
     close () {
       this.entryLabels = this.selectedLabels.map(this.labelToItem)
       this.$emit("input", false);

lib/www/server/events/handlers/detect-fdsp.js

@@ -69,7 +69,7 @@ class DetectFDSP {
 								point: prev._point,
 								remarks: "Last shotpoint of the day",
 								labels: ["LDSP", "Prod"],
-								meta: {auto: true, insertedBy: this.constructor.name}
+								meta: {auto: true, author: `*${this.constructor.name}*`}
 							};
 
 							const fdsp = {
@@ -77,7 +77,7 @@ class DetectFDSP {
 								point: cur._point,
 								remarks: "First shotpoint of the day",
 								labels: ["FDSP", "Prod"],
-								meta: {auto: true, insertedBy: this.constructor.name}
+								meta: {auto: true, author: `*${this.constructor.name}*`}
 							};
 
 							INFO("LDSP", ldsp);

lib/www/server/events/handlers/detect-soft-start.js

@@ -0,0 +1,128 @@
+const { schema2pid } = require('../../lib/db/connection');
+const { event } = require('../../lib/db');
+const { ALERT, ERROR, WARNING, NOTICE, INFO, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);
+
+class DetectSoftStart {
+	/* Data may come much faster than we can process it, so we put it
+	 * in a queue and process it at our own pace.
+	 *
+	 * The run() method fills the queue with the necessary data and then
+	 * calls processQueue().
+	 *
+	 * The processQueue() method looks takes the first two elements in
+	 * the queue and processes them if they are not already being taken
+	 * care of by a previous processQueue() call – this will happen when
+	 * data is coming in faster than it can be processed.
+	 *
+	 * If the processQueue() call is the first to see the two bottommost
+	 * two elements, it will process them and, when finished, it will set
+	 * the `isPending` flag of the bottommost element to `false`, thus
+	 * letting the next call know that it has work to do.
+	 *
+	 * If the queue was empty, run() will set the `isPending` flag of its
+	 * first element to a falsy value, thus bootstrapping the process.
+	 */
+	static MAX_QUEUE_SIZE = 125000;
+
+	queue = [];
+
+	async processQueue () {
+		DEBUG("Queue length", this.queue.length)
+		while (this.queue.length > 1) {
+			if (this.queue[0].isPending) {
+				DEBUG("Queue busy");
+				setImmediate(() => this.processQueue());
+				return;
+			}
+
+			const prev = this.queue.shift();
+			const cur  = this.queue[0];
+
+			try {
+				// DEBUG("Previous", prev);
+				// DEBUG("Current", cur);
+
+				// TODO:
+				// Consider whether to remember if soft start / full volume events
+				// have already been emitted and wait until there is an online/offline
+				// transition before re-emitting.
+				// This may or may not be a good idea.
+
+				// Look for a soft start or full volume event
+				if (cur.num_active >= 1 && !prev.num_active && cur.num_active < cur.num_guns) {
+					INFO("Soft start detected @", cur.tstamp);
+
+					const projectId = await schema2pid(cur._schema ?? prev._schema);
+
+					// TODO: Try and grab the corresponding comment from the configuration?
+					const payload = {
+						tstamp: cur.tstamp,
+						remarks: "Soft start",
+						labels: [ "Daily", "Guns", "Prod" ],
+						meta: {auto: true, author: `*${this.constructor.name}*`}
+					};
+					DEBUG("Posting event", projectId, payload);
+					await event.post(projectId, payload);
+
+				} else if (cur.num_active == cur.num_guns && prev.num_active < cur.num_active) {
+					INFO("Full volume detected @", cur.tstamp);
+
+					const projectId = await schema2pid(cur._schema ?? prev._schema);
+
+					// TODO: Try and grab the corresponding comment from the configuration?
+					const payload = {
+						tstamp: cur.tstamp,
+						remarks: "Full volume",
+						labels: [ "Daily", "Guns", "Prod" ],
+						meta: {auto: true, author: `*${this.constructor.name}*`}
+					};
+					DEBUG("Posting event", projectId, payload);
+					await event.post(projectId, payload);
+				}
+				// Processing of this shot has already been completed.
+				// The queue can now move forward.
+			} catch (err) {
+				ERROR("DetectSoftStart Error")
+				ERROR(err);
+			} finally {
+				cur.isPending = false;
+			}
+		}
+	}
+
+	async run (data) {
+		if (!data || data.channel !== "realtime") {
+			return;
+		}
+
+		if (!(data.payload && data.payload.new && data.payload.new.meta)) {
+			return;
+		}
+
+		const meta = data.payload.new.meta;
+
+		if (this.queue.length < DetectSoftStart.MAX_QUEUE_SIZE) {
+
+			this.queue.push({
+				isPending: this.queue.length,
+				_schema: meta._schema,
+				tstamp: meta.tstamp ?? meta.time,
+				shot: meta.shot,
+				lineStatus: meta.lineStatus,
+				_sequence: meta._sequence,
+				_point: meta._point,
+				lineName: meta.lineName,
+				num_guns: meta.num_guns,
+				num_active: meta.num_active
+			});
+
+		} else {
+			// FIXME Change to alert
+			ALERT("DetectSoftStart queue full at", this.queue.length);
+		}
+
+		this.processQueue();
+	}
+}
+
+module.exports = DetectSoftStart;

lib/www/server/events/handlers/detect-soleol.js

@@ -68,7 +68,8 @@ class DetectSOLEOL {
 							sequence,
 							point: cur._point,
 							remarks,
-							labels
+							labels,
+							meta: {auto: true, author: `*${this.constructor.name}*`}
 						}
 
 // 						console.log(projectId, payload);
@@ -101,7 +102,8 @@ class DetectSOLEOL {
 							sequence,
 							point: prev._point,
 							remarks,
-							labels
+							labels,
+							meta: {auto: true, author: `*${this.constructor.name}*`}
 						}
 
 // 						console.log(projectId, payload);

lib/www/server/events/handlers/index.js

@@ -1,5 +1,7 @@
 const Handlers = [
 	require('./detect-soleol'),
+	require('./detect-soft-start'),
+	require('./report-line-change-time'),
 	require('./detect-fdsp')
 ];
 

lib/www/server/events/handlers/report-line-change-time.js

@@ -0,0 +1,301 @@
+const { schema2pid } = require('../../lib/db/connection');
+const { event, project } = require('../../lib/db');
+const { withinValidity } = require('../../lib/utils/ranges');
+const unique = require('../../lib/utils/unique');
+const { ALERT, ERROR, WARNING, NOTICE, INFO, DEBUG } = require('DOUGAL_ROOT/debug')(__filename);
+
+class ReportLineChangeTime {
+	/* Data may come much faster than we can process it, so we put it
+	 * in a queue and process it at our own pace.
+	 *
+	 * The run() method fills the queue with the necessary data and then
+	 * calls processQueue().
+	 *
+	 * The processQueue() method looks takes the first two elements in
+	 * the queue and processes them if they are not already being taken
+	 * care of by a previous processQueue() call – this will happen when
+	 * data is coming in faster than it can be processed.
+	 *
+	 * If the processQueue() call is the first to see the two bottommost
+	 * two elements, it will process them and, when finished, it will set
+	 * the `isPending` flag of the bottommost element to `false`, thus
+	 * letting the next call know that it has work to do.
+	 *
+	 * If the queue was empty, run() will set the `isPending` flag of its
+	 * first element to a falsy value, thus bootstrapping the process.
+	 */
+	static MAX_QUEUE_SIZE = 125000;
+
+	queue = [];
+
+	author = `*${this.constructor.name}*`;
+
+	async processQueue () {
+		DEBUG("Queue length", this.queue.length)
+		while (this.queue.length > 0) {
+			if (this.queue[0].isPending) {
+				DEBUG("Queue busy");
+				setImmediate(() => this.processQueue());
+				return;
+			}
+
+			const cur = this.queue.shift();
+			const next = this.queue[0];
+			const projectId = cur.pid;
+			// Are we being called because of a LGSP or because of a FGSP?
+			const forward = (cur.old?.labels?.includes("LGSP") || cur.new?.labels?.includes("LGSP"));
+
+			if (!projectId) {
+				WARNING("No projectID found in event", cur);
+				return;
+			}
+
+			async function getConfiguration (projectId) {
+				return await project.configuration.get(projectId);
+			}
+
+
+			async function getLineChangeTime (data, forward = false) {
+				if (forward) {
+					const ospEvents = await event.list(projectId, {label: "FGSP"});
+					// DEBUG("ospEvents", ospEvents);
+					const osp = ospEvents.filter(i => i.tstamp > data.tstamp).pop();
+					DEBUG("fsp", osp);
+					// DEBUG("data", data);
+
+					if (osp) {
+						DEBUG("lineChangeTime", osp.tstamp - data.tstamp);
+						return { lineChangeTime: osp.tstamp - data.tstamp, osp };
+					}
+				} else {
+					const ospEvents = await event.list(projectId, {label: "LGSP"});
+					// DEBUG("ospEvents", ospEvents);
+					const osp = ospEvents.filter(i => i.tstamp < data.tstamp).shift();
+					DEBUG("lsp", osp);
+					// DEBUG("data", data);
+
+					if (osp) {
+						DEBUG("lineChangeTime", data.tstamp - osp.tstamp);
+						return { lineChangeTime: data.tstamp - osp.tstamp, osp };
+					}
+				}
+			}
+
+			function parseInterval (dt) {
+				const daySeconds = (dt/1000) % 86400;
+				const d = Math.floor((dt/1000) / 86400);
+				const dateObject = new Date(null);
+				dateObject.setSeconds(daySeconds);
+				const [ h, m, s ] = dateObject.toISOString().slice(11, 19).split(":").map(Number);
+				return {d, h, m, s};
+			}
+
+			function formatInterval (i) {
+				let str = "";
+				for (let [k, v] of Object.entries(i)) {
+					if (v) {
+						str += " " + v + " " + k;
+					}
+				}
+				return str.trim();
+			}
+
+			const deleteStaleEvents = async (seq) => {
+				if (seq) {
+					DEBUG("Will delete lct events related to sequence(s)", seq);
+
+					const jpq = `$."${this.author}"`;
+
+					const opts = {jpq};
+
+					if (Array.isArray(seq)) {
+						opts.sequences = unique(seq).filter(i => !!i);
+					} else {
+						opts.sequence = seq;
+					}
+
+					const staleEvents = await event.list(projectId, opts);
+					DEBUG(staleEvents.length ?? 0, "events to delete");
+					for (let staleEvent of staleEvents) {
+						DEBUG(`Deleting event id ${staleEvent.id} (seq = ${staleEvent.sequence}, point = ${staleEvent.point})`);
+						await event.del(projectId, staleEvent.id);
+					}
+				}
+			}
+
+			const createLineChangeTimeEvents = async (lineChangeTime, data, osp) => {
+
+				const events = [];
+				const cfg = (await project.configuration.get(projectId));
+				const nlcd = cfg?.production?.nominalLineChangeDuration * 60*1000; // m → ms
+				DEBUG("nlcd", nlcd);
+				if (nlcd && lineChangeTime > nlcd) {
+					const excess = lineChangeTime-nlcd;
+					const excessString = formatInterval(parseInterval(excess));
+					DEBUG("excess", excess, excessString);
+
+					// ref: The later of the two events
+					const ref = forward ? osp : data;
+					const payload = {
+						// tstamp: new Date(ref.tstamp-1),
+						sequence: ref.sequence,
+						point: ref.point,
+						remarks: `_Nominal line change duration exceeded by ${excessString}_`,
+						labels: [ "Nav", "Prod" ],
+						meta: {
+							auto: true,
+							author: this.author,
+							[this.author]: {
+								parents: [
+									data.id,
+									osp.id
+								],
+								type: "excess",
+								value: excess
+							}
+						}
+					}
+
+					events.push(payload);
+					DEBUG("Created line change duration exceeded event", projectId, payload);
+				}
+
+
+				const lctString = formatInterval(parseInterval(lineChangeTime));
+
+				// ref: The later of the two events
+				const ref = forward ? osp : data;
+				const payload = {
+					// tstamp: new Date(ref.tstamp-1),
+					sequence: ref.sequence,
+					point: ref.point,
+					remarks: `Line change time: ${lctString}`,
+					labels: [ "Nav", "Prod" ],
+					meta: {
+						auto: true,
+						author: this.author,
+						[this.author]: {
+							parents: [
+								data.id,
+								osp.id
+							],
+							type: "lineChangeTime",
+							value: lineChangeTime
+						}
+					}
+				};
+
+				events.push(payload);
+				DEBUG("Created line change duration event", projectId, payload);
+
+				return events;
+			}
+
+			const maybePostEvent = async (projectId, payload) => {
+				DEBUG("Posting event", projectId, payload);
+				await event.post(projectId, payload);
+			}
+
+			try {
+				// DEBUG("Previous", prev);
+				DEBUG("Current", cur);
+				DEBUG("Forward search", forward);
+
+				// We have these scenarios to consider:
+				// INSERT:
+				//   `old` will be NULL
+				//   Add event with line change time:
+				//     - match validity with `new`
+				//     - meta.ReportLineChangeTime.link  refers to new.uid (or new.id?)
+				// UPDATE:
+				//   `old` is not NULL
+				//   `new` is not NULL
+				//   Delete previous event from event_log (not event_log_full)
+				//   Add event with line change time:
+				//     - match validity with `new`
+				//     - meta.ReportLineChangeTime.link  refers to new.uid (or new.id?)
+				// DELETE:
+				//   `old` is not NULL
+				//   `new` will be NULL
+				//   Delete previous event from event_log (not event_log_full)
+
+				await deleteStaleEvents([cur.old?.sequence, cur.new?.sequence]);
+
+				if (cur.operation == "INSERT") {
+					// NOTE: UPDATE on the event_log view translates to one UPDATE plus one INSERT
+					// on event_log_full, so we don't need to worry about UPDATE here.
+					const data = cur.new;
+					DEBUG("INSERT seen: will add lct events related to ", data.id);
+
+					if (withinValidity(data.validity)) {
+						DEBUG("Event within validity period", data.validity, new Date());
+
+						data.tstamp = new Date(data.tstamp);
+						const { lineChangeTime, osp } = await getLineChangeTime(data, forward);
+
+						if (lineChangeTime) {
+
+							const events = await createLineChangeTimeEvents(lineChangeTime, data, osp);
+
+							if (events?.length) {
+								DEBUG("Deleting other events for sequence", events[0].sequence);
+								await deleteStaleEvents(events[0].sequence);
+							}
+
+							for (let payload of events) {
+								await maybePostEvent(projectId, payload);
+							}
+						}
+					} else {
+						DEBUG("Event outside of validity range", data.validity, "lct events not inserted");
+					}
+
+				}
+
+
+				// Processing of this shot has already been completed.
+				// The queue can now move forward.
+			} catch (err) {
+				ERROR("ReportLineChangeTime Error")
+				ERROR(err);
+			} finally {
+				if (next) {
+					next.isPending = false;
+				}
+			}
+		}
+	}
+
+	async run (data) {
+		if (!data || data.channel !== "event") {
+			return;
+		}
+
+		if (!(data.payload?.new?.labels) && !(data.payload?.old?.labels)) {
+			return;
+		}
+
+		const n = data.payload.new;
+		const o = data.payload.old;
+
+		if (!n?.labels?.includes("FGSP") && !o?.labels?.includes("FGSP") &&
+			!n?.labels?.includes("LGSP") && !o?.labels?.includes("LGSP")) {
+			return;
+		}
+
+		if (this.queue.length < ReportLineChangeTime.MAX_QUEUE_SIZE) {
+
+			this.queue.push({
+				...data.payload,
+				isPending: this.queue.length,
+			});
+
+		} else {
+			ALERT("ReportLineChangeTime queue full at", this.queue.length);
+		}
+
+		this.processQueue();
+	}
+}
+
+module.exports = ReportLineChangeTime;

lib/www/server/lib/db/event/list.js

@@ -10,25 +10,34 @@ async function list (projectId, opts = {}) {
 	const offset = Math.abs((opts.page-1)*opts.itemsPerPage) || 0;
 	const limit = Math.abs(Number(opts.itemsPerPage)) || null;
 
-	const filter = opts.sequence
-		? String(opts.sequence).includes(";")
-			? [ "sequence = ANY ( $1 )", [ opts.sequence.split(";") ] ]
-			: [ "sequence = $1", [ opts.sequence ] ]
-		: opts.date0
-			? opts.date1
-				? [ "date(tstamp) BETWEEN SYMMETRIC $1 AND $2", [ opts.date0, opts.date1 ] ]
-				: [ "date(tstamp) = $1", [ opts.date0 ] ]
-			: [ "true = true", [] ];
+	const sequence = opts.sequence && Number(opts.sequence) || null;
+	const sequences = opts.sequences && (Array.isArray(opts.sequences)
+		? opts.sequences.map(Number)
+		: opts.sequences.split(/[^0-9]+/).map(Number)) || null;
+	const date0 = opts.date0 ?? null;
+	const date1 = opts.date1 ?? null;
+	const jpq = opts.jpq || null;
+	const label = opts.label ?? null;
 
 	const text = `
 		SELECT *
 		FROM event_log e
 		WHERE
-			${filter[0]}
-		ORDER BY ${sortKey} ${sortDir};
+			($1::numeric IS NULL OR sequence = $1) AND
+			($2::numeric[] IS NULL OR sequence = ANY( $2 )) AND
+			($3::timestamptz IS NULL OR date(tstamp) = $3) AND
+			($3::timestamptz IS NULL OR
+				(($4::timestamptz IS NULL AND date(tstamp) = $3) OR
+					date(tstamp) BETWEEN SYMMETRIC $3 AND $4)) AND
+			($5::jsonpath IS NULL OR jsonb_path_exists(meta::jsonb, $5::jsonpath)) AND
+			($6::text IS NULL OR $6 = ANY(labels))
+		ORDER BY ${sortKey} ${sortDir}
+		LIMIT ${limit};
 	`;
 
-	const res = await client.query(text, filter[1]);
+	const values = [ sequence, sequences, date0, date1, jpq, label ];
+
+	const res = await client.query(text, values);
 	client.release();
 	return res.rows.map(i => replaceMarkers(i));
 }

lib/www/server/lib/db/event/post.js

@@ -9,10 +9,10 @@ async function post (projectId, payload, opts = {}) {
 
 		const text = `
 			INSERT
-			INTO event_log (tstamp, sequence, point, remarks, labels)
-			VALUES ($1, $2, $3, replace_placeholders($4, $1, $2, $3), $5);
+			INTO event_log (tstamp, sequence, point, remarks, labels, meta)
+			VALUES ($1, $2, $3, replace_placeholders($4, $1, $2, $3), $5, $6);
 		`;
-		const values = [ p.tstamp, p.sequence, p.point, p.remarks, p.labels ];
+		const values = [ p.tstamp, p.sequence, p.point, p.remarks, p.labels, p.meta ];
 
 		DEBUG("Inserting new values: %O", values);
 		await client.query(text, values);

lib/www/server/lib/db/navdata/save.js

@@ -290,12 +290,13 @@ async function save (navData, opts = {}) {
 		}
 	} else {
 		// We are offline. We only assign _schema once every save_interval seconds at most
+		// unless there is gun data present.
 		if (opts.offline_survey_heuristics == "nearest_preplot") {
 			const now = Date.now();
 			const do_save = !opts.offline_survey_detect_interval ||
 				(now - last_tstamp) >= opts.offline_survey_detect_interval;
 
-			if (do_save) {
+			if (do_save || "guns" in navData?.payload) {
 				const configs = await getAllProjectConfigs();
 				const candidates = await getCandidates(navData);
 				const bestCandidate = await getNearestOfflinePreplot(candidates);

lib/www/server/lib/utils/index.js

@@ -6,5 +6,7 @@ module.exports = {
 	flattenQCDefinitions: require('./flattenQCDefinitions'),
 	deepMerge: require('./deepMerge'),
 	removeNulls: require('./removeNulls'),
-	logicalPath: require('./logicalPath')
+	logicalPath: require('./logicalPath'),
+	ranges: require('./ranges'),
+	unique: require('./unique')
 };

lib/www/server/lib/utils/ranges.js

@@ -0,0 +1,74 @@
+
+function parseRange (str) {
+	const rx = /^[\[(].*,.*[)\]]$/
+
+	if (rx.test(str)) {
+		const lower_inclusive = str[0] == '[';
+		const upper_inclusive = str[str.length-1] == ']';
+		const [ lower, upper ] = str.slice(1,-1).split(",");
+		return {
+			upper,
+			lower,
+			upper_inclusive,
+			lower_inclusive
+		};
+	}
+}
+
+function parseValidity (str) {
+	const range = parseRange(str);
+
+	if (range) {
+		ts0 = range.lower ? new Date(range.lower) : null;
+		ts1 = range.upper ? new Date(range.upper) : null;
+
+		return {
+			...range,
+			lower: ts0,
+			upper: ts1
+		};
+	}
+}
+
+function withinValidity (range, ts) {
+	if (!ts) {
+		ts = new Date();
+	}
+
+	if (typeof range === "string") {
+		range = parseValidity(range);
+	}
+
+	if (range.lower) {
+		if (range.lower_inclusive) {
+			if (!(range.lower <= ts)) {
+				return false;
+			}
+		} else {
+			if (!(range.lower < ts)) {
+				return false;
+			}
+		}
+	}
+
+	if (range.upper) {
+		if (range.upper_inclusive) {
+			if (!(range.upper >= ts)) {
+				return false;
+			}
+		} else {
+			if (!(range.upper > ts)) {
+				return false;
+			}
+		}
+	}
+
+	return true;
+}
+
+module.exports = {
+	parseRange,
+	parseValidity,
+	withinValidity
+}
+

lib/www/server/lib/utils/unique.js

@@ -0,0 +1,6 @@
+
+function unique(array) {
+	return [...new Set(array)];
+}
+
+module.exports = unique;