Refactor flow model. Include buffer

Author: altavir

controls-constructor/src/commonMain/kotlin/space/kscience/controls/constructor/ConstructorElement.kt

@@ -83,7 +83,7 @@ public interface StateContainer : ContextAware, CoroutineScope {
     }
 }
 
-public interface Model: StateContainer
+public interface Model : StateContainer
 
 /**
  * Run simulation using context simulation dispatcher
@@ -96,7 +96,8 @@ public suspend fun <M : Model> M.runSimulation(
     }
 }
 
-public val StateContainer.states get() = constructorElements.filterIsInstance<StateConstructorElement<*>>().map { it.state }
+public val StateContainer.states
+    get() = constructorElements.filterIsInstance<StateConstructorElement<*>>().map { it.state }
 
 /**
  * Register a [state] in this container. The state is not registered as a device property if [this] is a [DeviceConstructor]
@@ -122,7 +123,7 @@ public fun <T : ModelConstructor> StateContainer.model(model: T): T {
  * Create and register a timer state.
  */
 public fun StateContainer.timer(tick: Duration): TimerState =
-    registerState(TimerState(context.request(ClockManager), tick))
+    registerState(TimerState(context.plugins[ClockManager] ?: context.request(ClockManager), tick))
 
 /**
  * Register a new timer and perform [block] on its change
@@ -181,6 +182,14 @@ public fun <T1, T2, R> StateContainer.combineState(
     transformation: (T1, T2) -> R,
 ): DeviceState<R> = registerState(DeviceState.combine(first, second, transformation))
 
+
+public fun <T1, T2, T3, R> StateContainer.combineState(
+    first: DeviceState<T1>,
+    second: DeviceState<T2>,
+    third: DeviceState<T3>,
+    transformation: (T1, T2, T3) -> R,
+): DeviceState<R> = registerState(DeviceState.combine(first, second, third, transformation))
+
 /**
  * Combines multiple device states into a single state by applying a transformation function.
  *

controls-constructor/src/commonMain/kotlin/space/kscience/controls/constructor/TimerState.kt

@@ -23,18 +23,18 @@ public class TimerState(
     initialValue: Instant = Instant.DISTANT_PAST,
 ) : DeviceState<Instant> {
 
-    private val clock = MutableStateFlow(initialValue)
+    private val time = MutableStateFlow(initialValue)
 
     private val updateJob = clockManager.context.launch(clockManager.simulationDispatcher) {
         while (isActive) {
-            clock.value = clockManager.clock.now()
+            time.value = clockManager.clock.now()
             delay(tick)
         }
     }
 
-    override val valueFlow: Flow<Instant> get() = clock
+    override val valueFlow: Flow<Instant> get() = time
 
-    override val value: Instant get() = clock.value
+    override val value: Instant get() = time.value
 
     override fun toString(): String = "TimerState(tick=$tick)"
 }

controls-constructor/src/commonMain/kotlin/space/kscience/controls/constructor/devices/Drive.kt

@@ -5,7 +5,7 @@ import space.kscience.controls.constructor.MutableDeviceState
 import space.kscience.controls.constructor.property
 import space.kscience.controls.constructor.units.NewtonsMeters
 import space.kscience.controls.constructor.units.Numeric
-import space.kscience.controls.constructor.units.numericalValue
+import space.kscience.controls.constructor.units.numeric
 import space.kscience.dataforge.context.Context
 import space.kscience.dataforge.meta.MetaConverter
 
@@ -15,5 +15,5 @@ public class Drive(
     context: Context,
     force: MutableDeviceState<Numeric<NewtonsMeters>> = MutableDeviceState(Numeric(0)),
 ) : DeviceConstructor(context) {
-    public val force: MutableDeviceState<Numeric<NewtonsMeters>> by property(MetaConverter.numericalValue(), force)
+    public val force: MutableDeviceState<Numeric<NewtonsMeters>> by property(MetaConverter.numeric(), force)
 }

controls-constructor/src/commonMain/kotlin/space/kscience/controls/constructor/devices/EncoderDevice.kt

@@ -5,7 +5,7 @@ import space.kscience.controls.constructor.DeviceState
 import space.kscience.controls.constructor.property
 import space.kscience.controls.constructor.units.Degrees
 import space.kscience.controls.constructor.units.Numeric
-import space.kscience.controls.constructor.units.numericalValue
+import space.kscience.controls.constructor.units.numeric
 import space.kscience.dataforge.context.Context
 import space.kscience.dataforge.meta.MetaConverter
 
@@ -16,5 +16,5 @@ public class EncoderDevice(
     context: Context,
     position: DeviceState<Numeric<Degrees>>
 ) : DeviceConstructor(context) {
-    public val position: DeviceState<Numeric<Degrees>> by property(MetaConverter.numericalValue<Degrees>(), position)
+    public val position: DeviceState<Numeric<Degrees>> by property(MetaConverter.numeric<Degrees>(), position)
 }

controls-constructor/src/commonMain/kotlin/space/kscience/controls/constructor/devices/LinearDrive.kt

@@ -6,7 +6,7 @@ import space.kscience.controls.constructor.models.PidRegulator
 import space.kscience.controls.constructor.units.Meters
 import space.kscience.controls.constructor.units.NewtonsMeters
 import space.kscience.controls.constructor.units.Numeric
-import space.kscience.controls.constructor.units.numericalValue
+import space.kscience.controls.constructor.units.numeric
 import space.kscience.dataforge.context.Context
 import space.kscience.dataforge.meta.Meta
 import space.kscience.dataforge.meta.MetaConverter
@@ -21,7 +21,7 @@ public class LinearDrive(
     meta: Meta = Meta.EMPTY,
 ) : DeviceConstructor(context, meta) {
 
-    public val position: DeviceState<Numeric<Meters>> by property(MetaConverter.numericalValue(), position)
+    public val position: DeviceState<Numeric<Meters>> by property(MetaConverter.numeric(), position)
 
     public val drive: Drive by device(drive)
     public val pid: PidRegulator<Meters, NewtonsMeters>  = model(

controls-constructor/src/commonMain/kotlin/space/kscience/controls/constructor/models/flow/Connections.kt

@@ -1,8 +1,124 @@
 package space.kscience.controls.constructor.models.flow
 
-//public class ContinuousBuffer(    context: Context,
-//                                  public val algebra: AmountAlgebra<T>,
-//                                  public val outputNames: Collection<String>,
-//                                  private val joinManagementStrategy: JoinManagementStrategy = JoinManagementStrategy.PROPORTIONAL,
-//) : ModelConstructor(context), ContinuousProducerModel<T> {
-//}
+import kotlinx.coroutines.delay
+import space.kscience.controls.constructor.*
+import space.kscience.controls.constructor.units.*
+import space.kscience.dataforge.context.Context
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
+
+/**
+ * A simulation model that represents a continuous buffer for storing a quantity of a measurable unit.
+ * The buffer dynamically responds to supply and consumption requests while ensuring constraints such as buffer capacity.
+ *
+ * @param U The unit of measurement associated with the buffer contents.
+ * @param T The amount type representing the measurable quantity stored in the buffer.
+ * @param context The simulation context in which the buffer operates.
+ * @param algebra The algebraic operations to manipulate the amount type.
+ * @param bufferCapacity The maximum capacity of the buffer as a device state.
+ * @param supplyRequest A device state representing the amount requested to be added to the buffer.
+ * @param consumerRequest A device state representing the amount requested to be consumed from the buffer.
+ * @param initialLevel The initial quantity stored in the buffer.
+ * @param timeStep The simulation time step for updating the buffer's state.
+ *
+ * The model tracks the current level of stored content, calculates remaining buffer space, and manages production and consumption dynamics.
+ * It ensures the buffer's level stays within the bounds defined by its capacity.
+ */
+public class ContinuousBuffer<U : UnitsOfMeasurement, T : Amount<U>>(
+    context: Context,
+    public val algebra: AmountAlgebra<U, T>,
+    public val bufferCapacity: DeviceState<Numeric<U>>,
+    override val supplyRequest: LateBindDeviceState<T> = LateBindDeviceState(algebra.zero),
+    override val consumerRequest: LateBindDeviceState<Numeric<U>> = LateBindDeviceState(Numeric.zero()),
+    initialLevel: T = algebra.zero,
+    timeStep: Duration = 1.seconds
+) : ModelConstructor(context), ContinuousProducerInterface<U, T>, ContinuousConsumerInterface<U, T> {
+
+    private val _content: MutableDeviceState<T> = MutableDeviceState(initialLevel)
+
+    public val content: DeviceState<T> get() = _content
+
+    init {
+        registerState(bufferCapacity)
+        registerState(content)
+        registerState(supplyRequest)
+        registerState(consumerRequest)
+    }
+
+    public val remainingBufferSpace: DeviceState<Numeric<U>> = combineState(
+        bufferCapacity, content
+    ) { capacity: Numeric<U>, content: T ->
+        capacity - Numeric<U>(content.value)
+    }
+
+    override val productionCapacity: DeviceState<T> = combineState(
+        supplyRequest,
+        content
+    ) { supply: T, content: T ->
+        with(algebra) {
+            supply + content
+        }
+    }
+
+    override val production: DeviceState<T> = combineState(
+        productionCapacity,
+        consumerRequest
+    ) { capacity, request ->
+        with(algebra) {
+            capacity.coerceValueIn(Numeric.zero<U>()..request)
+        }
+    }
+
+    override val consumationCapacity: DeviceState<Numeric<U>> = combineState(
+        remainingBufferSpace,
+        consumerRequest
+    ) { remaining: Numeric<U>, consumation: Numeric<U> ->
+        remaining + consumation
+    }
+
+    override val consumation: DeviceState<T> = combineState(
+        supplyRequest,
+        consumationCapacity
+    ) { supply: T, capacity ->
+        with(algebra) {
+            supply.coerceValueIn(Numeric.zero<U>()..capacity)
+        }
+    }
+
+    private val levelChange = onTimer(
+        tick = timeStep,
+        reads = listOf(production, consumation, remainingBufferSpace),
+        writes = listOf(content)
+    ) { prev, next ->
+        with(algebra) {
+            delay(timeStep)
+
+            val delta = consumation.value - production.value
+
+            _content.value = (_content.value + delta * (timeStep / 1.seconds))
+                .coerceValueIn(Numeric.zero<U>()..bufferCapacity.value)
+        }
+    }
+}
+
+/**
+ * Creates a [ContinuousBuffer] model that represents a buffer for storing and managing
+ * quantities of a measurable unit in a simulation context. The buffer ensures that its
+ * contents respect the given capacity constraints and dynamically manages supply requests.
+ *
+ * @param context The simulation context in which the buffer operates.
+ * @param capacity The maximum capacity of the buffer, represented as a [Numeric] value.
+ * @param supplyRequest A device state representing the amount requested to be supplied
+ *        to the buffer. Defaults to a [LateBindDeviceState] with a value of zero.
+ * @return A [ContinuousBuffer] instance configured with the specified parameters.
+ */
+public fun <U : UnitsOfMeasurement> ContinuousBuffer(
+    context: Context,
+    capacity: Numeric<U>,
+    supplyRequest: LateBindDeviceState<Numeric<U>> = LateBindDeviceState(Numeric(0))
+): ContinuousBuffer<U, Numeric<U>> = ContinuousBuffer(
+    context,
+    NumericAmountAlgebra<U>(),
+    DeviceState(capacity),
+    supplyRequest
+)

controls-constructor/src/commonMain/kotlin/space/kscience/controls/constructor/models/flow/ContinuousBuffer.kt

@@ -4,54 +4,61 @@ import space.kscience.controls.constructor.*
 import space.kscience.controls.constructor.units.*
 import space.kscience.dataforge.context.Context
 
+public interface ContinuousConsumerInterface<U: UnitsOfMeasurement, T : Amount<U>> {
+    public val consumation: DeviceState<T>
+    public val consumationCapacity: DeviceState<Numeric<U>>
+    public val supplyRequest: LateBindDeviceState<T>
+
+}
+
+public fun <U : UnitsOfMeasurement, T : Amount<U>> ContinuousConsumerInterface<U, T>.connectProducer(
+    producerCapacity: DeviceState<T>,
+) {
+    supplyRequest.bind(producerCapacity)
+}
+
 /**
  * Represents a model for a material flow consumer capable of consuming material flow based on its defined capacity
  * and requested supply. This class calculates the actual material flow consumed and the efficiency of consumption.
  *
  * @param U The type of units of measurement for the material flow.
  * @param context The execution context used for state management and operations.
- * @param capacity The maximum capacity for material flow consumption of the consumer.
+ * @param consumationCapacity The maximum capacity for material flow consumption of the consumer.
  * @param supplyRequest The state representing the requested material flow to be supplied.
  *
  * @property consumation A device state representing the actual material flow consumed,
  * calculated as the minimum of the requested supply and the consumer's capacity.
  * @property efficiency A device state representing the efficiency of the consumer, calculated
  * as the ratio of the actual consumption to the capacity.
  */
-public class ContinuousConsumer<T : Amount<*>>(
+public class ContinuousConsumer<U: UnitsOfMeasurement, T : Amount<U>>(
     context: Context,
-    public val algebra: AmountAlgebra<T>,
-    public val capacity: DeviceState<T>,
-    public val supplyRequest: LateBindDeviceState<T> = LateBindDeviceState(algebra.zero)
-) : ModelConstructor(context) {
+    public val algebra: AmountAlgebra<U, T>,
+    override val consumationCapacity: DeviceState<Numeric<U>>,
+    override val supplyRequest: LateBindDeviceState<T> = LateBindDeviceState(algebra.zero)
+) : ModelConstructor(context), ContinuousConsumerInterface<U, T> {
 
     init {
-        registerState(capacity)
+        registerState(consumationCapacity)
         registerState(supplyRequest)
     }
 
-    public val consumation: DeviceState<T> = combineState(
+    override val consumation: DeviceState<T> = combineState(
         supplyRequest,
-        capacity
+        consumationCapacity
     ) { request, capacity ->
         with(algebra) {
-            minOf(request, capacity)
+            request.coerceValueIn(Numeric.zero<U>()..capacity)
         }
     }
 
     public val efficiency: DeviceState<Double> = combineState(
         consumation,
-        capacity
+        consumationCapacity
     ) { consumation, capacity ->
         consumation.value / capacity.value
     }
 
-    public fun connectProducer(
-        producerCapacity: DeviceState<T>,
-    ) {
-        this.supplyRequest.bind(producerCapacity)
-    }
-
     public companion object
 }
 
@@ -70,4 +77,4 @@ public fun <U : UnitsOfMeasurement> ContinuousConsumer(
     context: Context,
     capacity: DeviceState<Numeric<U>>,
     supplyRequest: LateBindDeviceState<Numeric<U>> = LateBindDeviceState(Numeric(0))
-): ContinuousConsumer<Numeric<U>> = ContinuousConsumer(context, NumericAmountAlgebra<U>(), capacity, supplyRequest)
+): ContinuousConsumer<U, Numeric<U>> = ContinuousConsumer(context, NumericAmountAlgebra<U>(), capacity, supplyRequest)

controls-constructor/src/commonMain/kotlin/space/kscience/controls/constructor/models/flow/ContinuousConsumer.kt

@@ -0,0 +1,36 @@
+package space.kscience.controls.constructor.models.flow
+
+import space.kscience.controls.constructor.DeviceState
+import space.kscience.controls.constructor.LateBindDeviceState
+import space.kscience.controls.constructor.ModelConstructor
+import space.kscience.controls.constructor.model
+import space.kscience.controls.constructor.units.Amount
+import space.kscience.controls.constructor.units.Numeric
+import space.kscience.controls.constructor.units.UnitsOfMeasurement
+import space.kscience.dataforge.context.Context
+
+public abstract class ContinuousFlowModel(
+    context: Context,
+    vararg dependencies: DeviceState<*>
+) : ModelConstructor(context, *dependencies) {
+
+    public companion object {
+        public fun <U : UnitsOfMeasurement, T : Amount<U>> connect(
+            producer: ContinuousProducerInterface<U, T>,
+            consumer: ContinuousConsumerInterface<U, T>,
+        ) {
+            producer.connectConsumer(consumer.consumationCapacity)
+            consumer.connectProducer(producer.productionCapacity)
+        }
+    }
+}
+
+public fun <U : UnitsOfMeasurement> ContinuousFlowModel.producer(
+    capacity: DeviceState<Numeric<U>>,
+    supplyRequest: LateBindDeviceState<Numeric<U>> = LateBindDeviceState(Numeric(0))
+): ContinuousProducer<U, Numeric<U>> = model(ContinuousProducer(context, capacity, supplyRequest))
+
+public fun <U : UnitsOfMeasurement> ContinuousFlowModel.consumer(
+    capacity: DeviceState<Numeric<U>>,
+    supplyRequest: LateBindDeviceState<Numeric<U>> = LateBindDeviceState(Numeric(0))
+): ContinuousConsumer<U, Numeric<U>> = model(ContinuousConsumer(context, capacity, supplyRequest))