Add fixed age plots for properties and states.

controls-core/src/commonMain/kotlin/space/kscience/controls/misc/ValueWithTime.kt

@@ -0,0 +1,69 @@
+package space.kscience.controls.misc
+
+import io.ktor.utils.io.core.Input
+import io.ktor.utils.io.core.Output
+import kotlinx.datetime.Instant
+import space.kscience.dataforge.io.IOFormat
+import space.kscience.dataforge.meta.Meta
+import space.kscience.dataforge.meta.get
+import space.kscience.dataforge.meta.transformations.MetaConverter
+import kotlin.reflect.KType
+import kotlin.reflect.typeOf
+
+/**
+ * A value coupled to a time it was obtained at
+ */
+public data class ValueWithTime<T>(val value: T, val time: Instant) {
+    public companion object {
+        /**
+         * Create a [ValueWithTime] format for given value value [IOFormat]
+         */
+        public fun <T> ioFormat(
+            valueFormat: IOFormat<T>,
+        ): IOFormat<ValueWithTime<T>> = ValueWithTimeIOFormat(valueFormat)
+
+        /**
+         * Create a [MetaConverter] with time for given value [MetaConverter]
+         */
+        public fun <T> metaConverter(
+            valueConverter: MetaConverter<T>,
+        ): MetaConverter<ValueWithTime<T>> = ValueWithTimeMetaConverter(valueConverter)
+
+
+        public const val META_TIME_KEY: String = "time"
+        public const val META_VALUE_KEY: String = "value"
+    }
+}
+
+private class ValueWithTimeIOFormat<T>(val valueFormat: IOFormat<T>) : IOFormat<ValueWithTime<T>> {
+    override val type: KType get() = typeOf<ValueWithTime<T>>()
+
+    override fun readObject(input: Input): ValueWithTime<T> {
+        val timestamp = InstantIOFormat.readObject(input)
+        val value = valueFormat.readObject(input)
+        return ValueWithTime(value, timestamp)
+    }
+
+    override fun writeObject(output: Output, obj: ValueWithTime<T>) {
+        InstantIOFormat.writeObject(output, obj.time)
+        valueFormat.writeObject(output, obj.value)
+    }
+
+}
+
+private class ValueWithTimeMetaConverter<T>(
+    val valueConverter: MetaConverter<T>,
+) : MetaConverter<ValueWithTime<T>> {
+    override fun metaToObject(
+        meta: Meta,
+    ): ValueWithTime<T>? = valueConverter.metaToObject(meta[ValueWithTime.META_VALUE_KEY] ?: Meta.EMPTY)?.let {
+        ValueWithTime(it, meta[ValueWithTime.META_TIME_KEY]?.instant ?: Instant.DISTANT_PAST)
+    }
+
+    override fun objectToMeta(obj: ValueWithTime<T>): Meta = Meta {
+        ValueWithTime.META_TIME_KEY put obj.time.toMeta()
+        ValueWithTime.META_VALUE_KEY put valueConverter.objectToMeta(obj.value)
+    }
+}
+
+public fun <T : Any> MetaConverter<T>.withTime(): MetaConverter<ValueWithTime<T>> = ValueWithTimeMetaConverter(this)

controls-core/src/commonMain/kotlin/space/kscience/controls/misc/timeIO.kt

@@ -0,0 +1,40 @@
+package space.kscience.controls.misc
+
+import io.ktor.utils.io.core.*
+import kotlinx.datetime.Instant
+import space.kscience.dataforge.context.Context
+import space.kscience.dataforge.io.IOFormat
+import space.kscience.dataforge.io.IOFormatFactory
+import space.kscience.dataforge.meta.Meta
+import space.kscience.dataforge.meta.string
+import space.kscience.dataforge.names.Name
+import space.kscience.dataforge.names.asName
+import kotlin.reflect.KType
+import kotlin.reflect.typeOf
+
+
+/**
+ * An [IOFormat] for [Instant]
+ */
+public object InstantIOFormat : IOFormat<Instant>, IOFormatFactory<Instant> {
+    override fun build(context: Context, meta: Meta): IOFormat<Instant> = this
+
+    override val name: Name = "instant".asName()
+
+    override val type: KType get() = typeOf<Instant>()
+
+    override fun writeObject(output: Output, obj: Instant) {
+        output.writeLong(obj.epochSeconds)
+        output.writeInt(obj.nanosecondsOfSecond)
+    }
+
+    override fun readObject(input: Input): Instant {
+        val seconds = input.readLong()
+        val nanoseconds = input.readInt()
+        return Instant.fromEpochSeconds(seconds, nanoseconds)
+    }
+}
+
+public fun Instant.toMeta(): Meta = Meta(toString())
+
+public val Meta.instant: Instant? get() = value?.string?.let { Instant.parse(it) }

controls-core/src/commonMain/kotlin/space/kscience/controls/misc/timeMeta.kt

@@ -1,18 +0,0 @@
-package space.kscience.controls.misc
-
-import kotlinx.datetime.Instant
-import space.kscience.dataforge.meta.Meta
-import space.kscience.dataforge.meta.get
-import space.kscience.dataforge.meta.long
-
-// TODO move to core
-
-public fun Instant.toMeta(): Meta = Meta {
-    "seconds" put epochSeconds
-    "nanos" put nanosecondsOfSecond
-}
-
-public fun Meta.instant(): Instant = value?.long?.let { Instant.fromEpochMilliseconds(it) } ?: Instant.fromEpochSeconds(
-    get("seconds")?.long ?: 0L,
-    get("nanos")?.long ?: 0L,
-)

controls-opcua/src/main/kotlin/space/kscience/controls/opcua/client/MetaBsdParser.kt

@@ -147,7 +147,7 @@ internal class MetaStructureCodec(
             "Float" -> member.value?.numberOrNull?.toFloat()
             "Double" -> member.value?.numberOrNull?.toDouble()
             "String" -> member.string
-            "DateTime" -> DateTime(member.instant().toJavaInstant())
+            "DateTime" -> DateTime(member.instant.toJavaInstant())
             "Guid" -> member.string?.let { UUID.fromString(it) }
             "ByteString" -> member.value?.list?.let { list ->
                 ByteString(list.map { it.number.toByte() }.toByteArray())

controls-vision/src/commonMain/kotlin/plotExtensions.kt

@@ -4,18 +4,27 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.transform
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.datetime.Clock
+import kotlinx.datetime.Instant
 import space.kscience.controls.api.Device
 import space.kscience.controls.api.propertyMessageFlow
 import space.kscience.controls.constructor.DeviceState
 import space.kscience.controls.manager.clock
+import space.kscience.controls.misc.ValueWithTime
 import space.kscience.dataforge.context.Context
 import space.kscience.dataforge.meta.*
 import space.kscience.plotly.Plot
 import space.kscience.plotly.bar
 import space.kscience.plotly.models.Bar
 import space.kscience.plotly.models.Scatter
+import space.kscience.plotly.models.Trace
 import space.kscience.plotly.models.TraceValues
 import space.kscience.plotly.scatter
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.hours
 
 private var TraceValues.values: List<Value>
     get() = value?.list ?: emptyList()
@@ -23,48 +32,126 @@ private var TraceValues.values: List<Value>
         value = ListValue(newValues)
     }
 
+
+private var TraceValues.times: List<Instant>
+    get() = value?.list?.map { Instant.parse(it.string) } ?: emptyList()
+    set(newValues) {
+        value = ListValue(newValues.map { it.toString().asValue() })
+    }
+
+
+private class TimeData(private var points: MutableList<ValueWithTime<Value>> = mutableListOf()) {
+    private val mutex = Mutex()
+
+    suspend fun append(time: Instant, value: Value) = mutex.withLock {
+        points.add(ValueWithTime(value, time))
+    }
+
+    suspend fun trim(maxAge: Duration, maxPoints: Int = 800, minPoints: Int = 400) {
+        require(maxPoints > 2)
+        require(minPoints > 0)
+        require(maxPoints > minPoints)
+        val now = Clock.System.now()
+        // filter old points
+        points.removeAll { now - it.time > maxAge }
+
+        if (points.size > maxPoints) {
+            val durationBetweenPoints = maxAge / minPoints
+            val markedForRemoval = buildList<ValueWithTime<Value>> {
+                var lastTime: Instant? = null
+                points.forEach { point ->
+                    if (lastTime?.let { point.time - it < durationBetweenPoints } == true) {
+                        add(point)
+                    } else {
+                        lastTime = point.time
+                    }
+                }
+            }
+            points.removeAll(markedForRemoval)
+        }
+    }
+
+    suspend fun fillPlot(x: TraceValues, y: TraceValues) = mutex.withLock {
+        x.strings = points.map { it.time.toString() }
+        y.values = points.map { it.value }
+    }
+}
+
 /**
- * Add a trace that shows a [Device] property change over time. Show only latest [pointsNumber] .
+ * Add a trace that shows a [Device] property change over time. Show only latest [maxPoints] .
  * @return a [Job] that handles the listener
  */
 public fun Plot.plotDeviceProperty(
     device: Device,
     propertyName: String,
     extractValue: Meta.() -> Value = { value ?: Null },
-    pointsNumber: Int = 400,
+    maxAge: Duration = 1.hours,
+    maxPoints: Int = 800,
+    minPoints: Int = 400,
     coroutineScope: CoroutineScope = device.context,
     configuration: Scatter.() -> Unit = {},
 ): Job = scatter(configuration).run {
     val clock = device.context.clock
-    device.propertyMessageFlow(propertyName).onEach { message ->
+    val data = TimeData()
-        x.strings = (x.strings + (message.time ?: clock.now()).toString()).takeLast(pointsNumber)
+    device.propertyMessageFlow(propertyName).transform {
-        y.values = (y.values + message.value.extractValue()).takeLast(pointsNumber)
+        data.append(it.time ?: clock.now(), it.value.extractValue())
+        data.trim(maxAge, maxPoints, minPoints)
+        emit(data)
+    }.onEach {
+        it.fillPlot(x, y)
     }.launchIn(coroutineScope)
 }
 
+private fun <T> Trace.updateFromState(
+    context: Context,
+    state: DeviceState<T>,
+    extractValue: T.() -> Value = { state.converter.objectToMeta(this).value ?: space.kscience.dataforge.meta.Null },
+    maxAge: Duration = 1.hours,
+    maxPoints: Int = 800,
+    minPoints: Int = 400,
+): Job{
+    val clock = context.clock
+    val data = TimeData()
+    return state.valueFlow.transform<T, TimeData> {
+        data.append(clock.now(), it.extractValue())
+        data.trim(maxAge, maxPoints, minPoints)
+    }.onEach {
+        it.fillPlot(x, y)
+    }.launchIn(context)
+}
+
+public fun <T> Plot.plotDeviceState(
+    context: Context,
+    state: DeviceState<T>,
+    extractValue: T.() -> Value = { state.converter.objectToMeta(this).value ?: Null },
+    maxAge: Duration = 1.hours,
+    maxPoints: Int = 800,
+    minPoints: Int = 400,
+    configuration: Scatter.() -> Unit = {},
+): Job = scatter(configuration).run {
+    updateFromState(context, state, extractValue, maxAge, maxPoints, minPoints)
+}
+
 
 public fun Plot.plotNumberState(
     context: Context,
     state: DeviceState<out Number>,
-    pointsNumber: Int = 400,
+    maxAge: Duration = 1.hours,
+    maxPoints: Int = 800,
+    minPoints: Int = 400,
     configuration: Scatter.() -> Unit = {},
 ): Job = scatter(configuration).run {
-    val clock = context.clock
+    updateFromState(context, state, { asValue() }, maxAge, maxPoints, minPoints)
-    state.valueFlow.onEach {
-        x.strings = (x.strings + clock.now().toString()).takeLast(pointsNumber)
-        y.numbers = (y.numbers + it).takeLast(pointsNumber)
-    }.launchIn(context)
 }
 
+
 public fun Plot.plotBooleanState(
     context: Context,
     state: DeviceState<Boolean>,
-    pointsNumber: Int = 400,
+    maxAge: Duration = 1.hours,
+    maxPoints: Int = 800,
+    minPoints: Int = 400,
     configuration: Bar.() -> Unit = {},
-): Job = bar(configuration).run {
+): Job =  bar(configuration).run {
-    val clock = context.clock
+    updateFromState(context, state, { asValue() }, maxAge, maxPoints, minPoints)
-    state.valueFlow.onEach {
-        x.strings = (x.strings + clock.now().toString()).takeLast(pointsNumber)
-        y.values = (y.values + it.asValue()).takeLast(pointsNumber)
-    }.launchIn(context)
 }

demo/constructor/src/jvmMain/kotlin/main.kt

@@ -51,32 +51,33 @@ public fun main() {
             val t = timeFromStart.toDouble(DurationUnit.SECONDS)
             val freq = 0.1
             val target = 5 * sin(2.0 * PI * freq * t) +
-                    sin(2 * PI * 21 * freq * t + 0.1 * (timeFromStart / pidParameters.timeStep))
+                    sin(2 * PI * 21 * freq * t + 0.02 * (timeFromStart / pidParameters.timeStep))
             pid.write(Regulator.target, target)
         }
     }
 
+    val maxAge = 10.seconds
 
     context.showDashboard {
         plot {
-            plotNumberState(context, state) {
+            plotNumberState(context, state, maxAge = maxAge) {
                 name = "real position"
             }
-            plotDeviceProperty(device["pid"], Regulator.position.name) {
+            plotDeviceProperty(device["pid"], Regulator.position.name, maxAge = maxAge) {
                 name = "read position"
             }
 
-            plotDeviceProperty(device["pid"], Regulator.target.name) {
+            plotDeviceProperty(device["pid"], Regulator.target.name, maxAge = maxAge) {
                 name = "target"
             }
         }
 
         plot {
-            plotDeviceProperty(device["start"], LimitSwitch.locked.name) {
+            plotDeviceProperty(device["start"], LimitSwitch.locked.name, maxAge = maxAge) {
                 name = "start measured"
                 mode = ScatterMode.markers
             }
-            plotDeviceProperty(device["end"], LimitSwitch.locked.name) {
+            plotDeviceProperty(device["end"], LimitSwitch.locked.name, maxAge = maxAge) {
                 name = "end measured"
                 mode = ScatterMode.markers
             }