Merge remote-tracking branch 'spc/dev' into dev

# Conflicts:
#	demo/all-things/build.gradle.kts

CHANGELOG.md

@@ -2,6 +2,28 @@
 
 ## Unreleased
 
+### Added
+
+### Changed
+
+### Deprecated
+
+### Removed
+
+### Fixed
+- Property writing does not trigger change if logical state already is the same as value to be set.
+- Modbus-slave triggers only once for multi-register write.
+
+### Security
+
+## 0.2.2-dev-1 - 2023-09-24
+
+### Changed
+- updating logical state in `DeviceBase` is now protected and called `propertyChanged()`
+- `DeviceBase` tries to read property after write if the writer does not set the value.
+
+## 0.2.1 - 2023-09-24
+
 ### Added
 - Core interfaces for building a device server
 - Magix service for binding controls devices (both as RPC client and server)
@@ -20,13 +42,3 @@
 - A magix event loop implementation in Kotlin. Includes HTTP/SSE and RSocket routes.
 - Magix history database API
 - ZMQ client endpoint for Magix
-
-### Changed
-
-### Deprecated
-
-### Removed
-
-### Fixed
-
-### Security

build.gradle.kts

@@ -14,7 +14,7 @@ val xodusVersion by extra("2.0.1")
 
 allprojects {
     group = "space.kscience"
-    version = "0.2.2-dev-1"
+    version = "0.2.2-dev-2"
     repositories{
         maven("https://maven.pkg.jetbrains.space/spc/p/sci/dev")
     }

controls-core/src/commonMain/kotlin/space/kscience/controls/api/Device.kt

@@ -124,5 +124,5 @@ public fun Device.getAllProperties(): Meta = Meta {
 /**
  * Subscribe on property changes for the whole device
  */
-public fun Device.onPropertyChange(callback: suspend PropertyChangedMessage.() -> Unit): Job =
+public fun Device.onPropertyChange(scope: CoroutineScope = this, callback: suspend PropertyChangedMessage.() -> Unit): Job =
-    messageFlow.filterIsInstance<PropertyChangedMessage>().onEach(callback).launchIn(this)
+    messageFlow.filterIsInstance<PropertyChangedMessage>().onEach(callback).launchIn(scope)

controls-core/src/commonMain/kotlin/space/kscience/controls/spec/DeviceBase.kt

@@ -1,24 +1,35 @@
 package space.kscience.controls.spec
 
-import kotlinx.coroutines.*
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.newCoroutineContext
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import space.kscience.controls.api.*
 import space.kscience.dataforge.context.Context
-import space.kscience.dataforge.context.error
+import space.kscience.dataforge.context.debug
 import space.kscience.dataforge.context.logger
 import space.kscience.dataforge.meta.Meta
+import space.kscience.dataforge.meta.get
+import space.kscience.dataforge.meta.int
 import space.kscience.dataforge.misc.DFExperimental
 import kotlin.coroutines.CoroutineContext
 
-
+/**
+ * Write a meta [item] to [device]
+ */
 @OptIn(InternalDeviceAPI::class)
 private suspend fun <D : Device, T> WritableDevicePropertySpec<D, T>.writeMeta(device: D, item: Meta) {
     write(device, converter.metaToObject(item) ?: error("Meta $item could not be read with $converter"))
 }
 
+/**
+ * Read Meta item from the [device]
+ */
 @OptIn(InternalDeviceAPI::class)
 private suspend fun <D : Device, T> DevicePropertySpec<D, T>.readMeta(device: D): Meta? =
     read(device)?.let(converter::objectToMeta)
@@ -39,7 +50,7 @@ private suspend fun <D : Device, I, O> DeviceActionSpec<D, I, O>.executeWithMeta
  */
 public abstract class DeviceBase<D : Device>(
     final override val context: Context,
-    override val meta: Meta = Meta.EMPTY,
+    final override val meta: Meta = Meta.EMPTY,
 ) : Device {
 
     /**
@@ -59,13 +70,7 @@ public abstract class DeviceBase<D : Device>(
         get() = actions.values.map { it.descriptor }
 
     override val coroutineContext: CoroutineContext by lazy {
-        context.newCoroutineContext(
+        context.newCoroutineContext(SupervisorJob(context.coroutineContext[Job]) + CoroutineName("Device $this"))
-            SupervisorJob(context.coroutineContext[Job]) +
-                    CoroutineName("Device $this") +
-                    CoroutineExceptionHandler { _, throwable ->
-                        logger.error(throwable) { "Exception in device $this job" }
-                    }
-        )
     }
 
 
@@ -74,7 +79,10 @@ public abstract class DeviceBase<D : Device>(
      */
     private val logicalState: HashMap<String, Meta?> = HashMap()
 
-    private val sharedMessageFlow: MutableSharedFlow<DeviceMessage> = MutableSharedFlow()
+    private val sharedMessageFlow: MutableSharedFlow<DeviceMessage> = MutableSharedFlow(
+        replay = meta["message.buffer"].int ?: 1000,
+        onBufferOverflow = BufferOverflow.DROP_OLDEST
+    )
 
     public override val messageFlow: SharedFlow<DeviceMessage> get() = sharedMessageFlow
 
@@ -87,7 +95,7 @@ public abstract class DeviceBase<D : Device>(
     /**
      * Update logical property state and notify listeners
      */
-    protected suspend fun updateLogical(propertyName: String, value: Meta?) {
+    protected suspend fun propertyChanged(propertyName: String, value: Meta?) {
         if (value != logicalState[propertyName]) {
             stateLock.withLock {
                 logicalState[propertyName] = value
@@ -99,10 +107,10 @@ public abstract class DeviceBase<D : Device>(
     }
 
     /**
-     * Update logical state using given [spec] and its convertor
+     * Notify the device that a property with [spec] value is changed
      */
-    public suspend fun <T> updateLogical(spec: DevicePropertySpec<D, T>, value: T) {
+    protected suspend fun <T> propertyChanged(spec: DevicePropertySpec<D, T>, value: T) {
-        updateLogical(spec.name, spec.converter.objectToMeta(value))
+        propertyChanged(spec.name, spec.converter.objectToMeta(value))
     }
 
     /**
@@ -112,7 +120,7 @@ public abstract class DeviceBase<D : Device>(
     override suspend fun readProperty(propertyName: String): Meta {
         val spec = properties[propertyName] ?: error("Property with name $propertyName not found")
         val meta = spec.readMeta(self) ?: error("Failed to read property $propertyName")
-        updateLogical(propertyName, meta)
+        propertyChanged(propertyName, meta)
         return meta
     }
 
@@ -122,7 +130,7 @@ public abstract class DeviceBase<D : Device>(
     public suspend fun readPropertyOrNull(propertyName: String): Meta? {
         val spec = properties[propertyName] ?: return null
         val meta = spec.readMeta(self) ?: return null
-        updateLogical(propertyName, meta)
+        propertyChanged(propertyName, meta)
         return meta
     }
 
@@ -135,15 +143,26 @@ public abstract class DeviceBase<D : Device>(
     }
 
     override suspend fun writeProperty(propertyName: String, value: Meta): Unit {
+        //bypass property setting if it already has that value
+        if (logicalState[propertyName] == value) {
+            logger.debug { "Skipping setting $propertyName to $value because value is already set" }
+            return
+        }
         when (val property = properties[propertyName]) {
             null -> {
-                //If there is a physical property with a given name, invalidate logical property and write physical one
+                //If there are no registered physical properties with given name, write a logical one.
-                updateLogical(propertyName, value)
+                propertyChanged(propertyName, value)
             }
 
             is WritableDevicePropertySpec -> {
+                //if there is a writeable property with a given name, invalidate logical and write physical
                 invalidate(propertyName)
                 property.writeMeta(self, value)
+                // perform read after writing if the writer did not set the value and the value is still in invalid state
+                if (logicalState[propertyName] == null) {
+                    val meta = property.readMeta(self)
+                    propertyChanged(propertyName, meta)
+                }
             }
 
             else -> {

controls-core/src/commonMain/kotlin/space/kscience/controls/spec/DevicePropertySpec.kt

@@ -115,6 +115,7 @@ public fun <D : Device, T> D.propertyFlow(spec: DevicePropertySpec<D, T>): Flow<
  */
 public fun <D : Device, T> D.onPropertyChange(
     spec: DevicePropertySpec<D, T>,
+    scope: CoroutineScope = this,
     callback: suspend PropertyChangedMessage.(T) -> Unit,
 ): Job = messageFlow
     .filterIsInstance<PropertyChangedMessage>()
@@ -124,15 +125,16 @@ public fun <D : Device, T> D.onPropertyChange(
         if (newValue != null) {
             change.callback(newValue)
         }
-    }.launchIn(this)
+    }.launchIn(scope)
 
 /**
  * Call [callback] on initial property value and each value change
  */
 public fun <D : Device, T> D.useProperty(
     spec: DevicePropertySpec<D, T>,
+    scope: CoroutineScope = this,
     callback: suspend (T) -> Unit,
-): Job = launch {
+): Job = scope.launch {
     callback(read(spec))
     messageFlow
         .filterIsInstance<PropertyChangedMessage>()

controls-core/src/commonMain/kotlin/space/kscience/controls/spec/DeviceSpec.kt

@@ -22,7 +22,7 @@ public val MetaConverter.Companion.unit: MetaConverter<Unit> get() = UnitMetaCon
 
 @OptIn(InternalDeviceAPI::class)
 public abstract class DeviceSpec<D : Device> {
-    //initializing meta property for everyone
+    //initializing the metadata property for everyone
     private val _properties = hashMapOf<String, DevicePropertySpec<D, *>>(
         DeviceMetaPropertySpec.name to DeviceMetaPropertySpec
     )

controls-modbus/src/main/kotlin/space/kscience/controls/modbus/DeviceProcessImage.kt

@@ -1,9 +1,9 @@
 package space.kscience.controls.modbus
 
 import com.ghgande.j2mod.modbus.procimg.*
-import io.ktor.utils.io.core.buildPacket
+import io.ktor.utils.io.core.*
-import io.ktor.utils.io.core.readByteBuffer
+import kotlinx.coroutines.delay
-import io.ktor.utils.io.core.writeShort
+import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import space.kscience.controls.api.Device
 import space.kscience.controls.spec.DevicePropertySpec
@@ -118,22 +118,48 @@ public class DeviceProcessImageBuilder<D : Device> internal constructor(
         }
     }
 
+    /**
+     * Trigger [block] if one of register changes.
+     */
+    private fun List<ObservableRegister>.onChange(block: (ByteReadPacket) -> Unit) {
+        var ready = false
+
+        forEach { register ->
+            register.addObserver { _, _ ->
+                ready = true
+            }
+        }
+
+        device.launch {
+            val builder = BytePacketBuilder()
+            while (isActive) {
+                delay(1)
+                if (ready) {
+                    val packet = builder.apply {
+                        forEach { value ->
+                            writeShort(value.toShort())
+                        }
+                    }.build()
+                    block(packet)
+                    ready = false
+                }
+            }
+        }
+    }
+
     public fun <T> bind(key: ModbusRegistryKey.HoldingRange<T>, propertySpec: WritableDevicePropertySpec<D, T>) {
         val registers = List(key.count) {
             ObservableRegister()
         }
+
         registers.forEachIndexed { index, register ->
-            register.addObserver { _, _ ->
-                val packet = buildPacket {
-                    registers.forEach { value ->
-                        writeShort(value.toShort())
-                    }
-                }
-                device[propertySpec] = key.format.readObject(packet)
-            }
             image.addRegister(key.address + index, register)
         }
 
+        registers.onChange { packet ->
+            device[propertySpec] = key.format.readObject(packet)
+        }
+
         device.useProperty(propertySpec) { value ->
             val packet = buildPacket {
                 key.format.writeObject(this, value)
@@ -182,20 +208,17 @@ public class DeviceProcessImageBuilder<D : Device> internal constructor(
         val registers = List(key.count) {
             ObservableRegister()
         }
+
         registers.forEachIndexed { index, register ->
-            register.addObserver { _, _ ->
-                val packet = buildPacket {
-                    registers.forEach { value ->
-                        writeShort(value.toShort())
-                    }
-                }
-                device.launch {
-                    device.action(key.format.readObject(packet))
-                }
-            }
             image.addRegister(key.address + index, register)
         }
 
+        registers.onChange { packet ->
+            device.launch {
+                device.action(key.format.readObject(packet))
+            }
+        }
+
         return registers
     }
 
@@ -205,11 +228,13 @@ public class DeviceProcessImageBuilder<D : Device> internal constructor(
  * Bind the device to Modbus slave (server) image.
  */
 public fun <D : Device> D.bindProcessImage(
+    unitId: Int = 0,
     openOnBind: Boolean = true,
     binding: DeviceProcessImageBuilder<D>.() -> Unit,
 ): ProcessImage {
-    val image = SimpleProcessImage()
+    val image = SimpleProcessImage(unitId)
     DeviceProcessImageBuilder(this, image).apply(binding)
+    image.setLocked(true)
     if (openOnBind) {
         launch {
             open()

controls-modbus/src/main/kotlin/space/kscience/controls/modbus/ModbusDevice.kt

@@ -21,9 +21,9 @@ import kotlin.reflect.KProperty
 public interface ModbusDevice : Device {
 
     /**
-     * Client id for this specific device
+     * Unit id for this specific device
      */
-    public val clientId: Int
+    public val unitId: Int
 
     /**
      * The modubus master connector
@@ -61,7 +61,7 @@ public interface ModbusDevice : Device {
     }
 
     public operator fun <T> ModbusRegistryKey.HoldingRange<T>.getValue(thisRef: Any?, property: KProperty<*>): T {
-        val packet = readInputRegistersToPacket(address, count)
+        val packet = readHoldingRegistersToPacket(address, count)
         return format.readObject(packet)
     }
 
@@ -82,35 +82,35 @@ public interface ModbusDevice : Device {
  * Read multiple sequential modbus coils (bit-values)
  */
 public fun ModbusDevice.readCoils(address: Int, count: Int): BitVector =
-    master.readCoils(clientId, address, count)
+    master.readCoils(unitId, address, count)
 
 public fun ModbusDevice.readCoil(address: Int): Boolean =
-    master.readCoils(clientId, address, 1).getBit(0)
+    master.readCoils(unitId, address, 1).getBit(0)
 
 public fun ModbusDevice.writeCoils(address: Int, values: BooleanArray) {
     val bitVector = BitVector(values.size)
     values.forEachIndexed { index, value ->
         bitVector.setBit(index, value)
     }
-    master.writeMultipleCoils(clientId, address, bitVector)
+    master.writeMultipleCoils(unitId, address, bitVector)
 }
 
 public fun ModbusDevice.writeCoil(address: Int, value: Boolean) {
-    master.writeCoil(clientId, address, value)
+    master.writeCoil(unitId, address, value)
 }
 
 public fun ModbusDevice.writeCoil(key: ModbusRegistryKey.Coil, value: Boolean) {
-    master.writeCoil(clientId, key.address, value)
+    master.writeCoil(unitId, key.address, value)
 }
 
 public fun ModbusDevice.readInputDiscretes(address: Int, count: Int): BitVector =
-    master.readInputDiscretes(clientId, address, count)
+    master.readInputDiscretes(unitId, address, count)
 
 public fun ModbusDevice.readInputDiscrete(address: Int): Boolean =
-    master.readInputDiscretes(clientId, address, 1).getBit(0)
+    master.readInputDiscretes(unitId, address, 1).getBit(0)
 
 public fun ModbusDevice.readInputRegisters(address: Int, count: Int): List<InputRegister> =
-    master.readInputRegisters(clientId, address, count).toList()
+    master.readInputRegisters(unitId, address, count).toList()
 
 private fun Array<out InputRegister>.toBuffer(): ByteBuffer {
     val buffer: ByteBuffer = ByteBuffer.allocate(size * 2)
@@ -129,10 +129,10 @@ private fun Array<out InputRegister>.toPacket(): ByteReadPacket = buildPacket {
 }
 
 public fun ModbusDevice.readInputRegistersToBuffer(address: Int, count: Int): ByteBuffer =
-    master.readInputRegisters(clientId, address, count).toBuffer()
+    master.readInputRegisters(unitId, address, count).toBuffer()
 
 public fun ModbusDevice.readInputRegistersToPacket(address: Int, count: Int): ByteReadPacket =
-    master.readInputRegisters(clientId, address, count).toPacket()
+    master.readInputRegisters(unitId, address, count).toPacket()
 
 public fun ModbusDevice.readDoubleInput(address: Int): Double =
     readInputRegistersToBuffer(address, Double.SIZE_BYTES).getDouble()
@@ -141,7 +141,7 @@ public fun ModbusDevice.readInputRegister(address: Int): Short =
     readInputRegisters(address, 1).first().toShort()
 
 public fun ModbusDevice.readHoldingRegisters(address: Int, count: Int): List<Register> =
-    master.readMultipleRegisters(clientId, address, count).toList()
+    master.readMultipleRegisters(unitId, address, count).toList()
 
 /**
  * Read a number of registers to a [ByteBuffer]
@@ -149,10 +149,10 @@ public fun ModbusDevice.readHoldingRegisters(address: Int, count: Int): List<Reg
  * @param count number of 2-bytes registers to read. Buffer size is 2*[count]
  */
 public fun ModbusDevice.readHoldingRegistersToBuffer(address: Int, count: Int): ByteBuffer =
-    master.readMultipleRegisters(clientId, address, count).toBuffer()
+    master.readMultipleRegisters(unitId, address, count).toBuffer()
 
 public fun ModbusDevice.readHoldingRegistersToPacket(address: Int, count: Int): ByteReadPacket =
-    master.readMultipleRegisters(clientId, address, count).toPacket()
+    master.readMultipleRegisters(unitId, address, count).toPacket()
 
 public fun ModbusDevice.readDoubleRegister(address: Int): Double =
     readHoldingRegistersToBuffer(address, Double.SIZE_BYTES).getDouble()
@@ -162,14 +162,14 @@ public fun ModbusDevice.readHoldingRegister(address: Int): Short =
 
 public fun ModbusDevice.writeHoldingRegisters(address: Int, values: ShortArray): Int =
     master.writeMultipleRegisters(
-        clientId,
+        unitId,
         address,
         Array<Register>(values.size) { SimpleInputRegister(values[it].toInt()) }
     )
 
 public fun ModbusDevice.writeHoldingRegister(address: Int, value: Short): Int =
     master.writeSingleRegister(
-        clientId,
+        unitId,
         address,
         SimpleInputRegister(value.toInt())
     )
@@ -183,10 +183,6 @@ public fun ModbusDevice.writeHoldingRegisters(address: Int, buffer: ByteBuffer):
     return writeHoldingRegisters(address, array)
 }
 
-public fun ModbusDevice.writeShortRegister(address: Int, value: Short) {
-    master.writeSingleRegister(address, SimpleInputRegister(value.toInt()))
-}
-
 public fun ModbusDevice.modbusRegister(
     address: Int,
 ): ReadWriteProperty<ModbusDevice, Short> = object : ReadWriteProperty<ModbusDevice, Short> {

controls-modbus/src/main/kotlin/space/kscience/controls/modbus/ModbusDeviceBySpec.kt

@@ -15,7 +15,7 @@ import space.kscience.dataforge.names.NameToken
 public open class ModbusDeviceBySpec<D: Device>(
     context: Context,
     spec: DeviceSpec<D>,
-    override val clientId: Int,
+    override val unitId: Int,
     override val master: AbstractModbusMaster,
     private val disposeMasterOnClose: Boolean = true,
     meta: Meta = Meta.EMPTY,

controls-modbus/src/main/kotlin/space/kscience/controls/modbus/ModbusRegistryMap.kt

@@ -152,7 +152,7 @@ public abstract class ModbusRegistryMap {
                     val rangeString = if (key.count == 1) {
                         key.address.toString()
                     } else {
-                        "${key.address} - ${key.address + key.count}"
+                        "${key.address} - ${key.address + key.count - 1}"
                     }
                     to.appendLine("${typeString}\t$rangeString\t$description")
                 }

demo/all-things/build.gradle.kts

@@ -12,7 +12,6 @@ repositories {
 
 val ktorVersion: String by rootProject.extra
 val rsocketVersion: String by rootProject.extra
-val visionforgeVersion: String by rootProject.extra
 
 dependencies {
     implementation(projects.controlsCore)
@@ -26,7 +25,6 @@ dependencies {
     implementation("io.ktor:ktor-client-cio:$ktorVersion")
     implementation("no.tornado:tornadofx:1.7.20")
     implementation("space.kscience:plotlykt-server:0.6.0")
-    implementation("space.kscience:visionforge-plotly:$visionforgeVersion")
 //    implementation("com.github.Ricky12Awesome:json-schema-serialization:0.6.6")
     implementation(spclibs.logback.classic)
 }

demo/mks-pdr900/src/main/kotlin/center/sciprog/devices/mks/MksPdr900Device.kt

@@ -49,16 +49,16 @@ class MksPdr900Device(context: Context, meta: Meta) : DeviceBySpec<MksPdr900Devi
         if (powerOnValue) {
             val ans = talk("FP!ON")
             if (ans == "ON") {
-                updateLogical(powerOn, true)
+                propertyChanged(powerOn, true)
             } else {
-                updateLogical(error, "Failed to set power state")
+                propertyChanged(error, "Failed to set power state")
             }
         } else {
             val ans = talk("FP!OFF")
             if (ans == "OFF") {
-                updateLogical(powerOn, false)
+                propertyChanged(powerOn, false)
             } else {
-                updateLogical(error, "Failed to set power state")
+                propertyChanged(error, "Failed to set power state")
             }
         }
     }
@@ -68,13 +68,13 @@ class MksPdr900Device(context: Context, meta: Meta) : DeviceBySpec<MksPdr900Devi
         invalidate(error)
         return if (answer.isNullOrEmpty()) {
             //            updateState(PortSensor.CONNECTED_STATE, false)
-            updateLogical(error, "No connection")
+            propertyChanged(error, "No connection")
             null
         } else {
             val res = answer.toDouble()
             if (res <= 0) {
-                updateLogical(powerOn, false)
+                propertyChanged(powerOn, false)
-                updateLogical(error, "No power")
+                propertyChanged(error, "No power")
                 null
             } else {
                 res

demo/motors/src/main/kotlin/ru/mipt/npm/devices/pimotionmaster/PiMotionMasterDevice.kt

@@ -172,7 +172,7 @@ class PiMotionMasterDevice(
             //Update port
             //address = portSpec.node
             port = portFactory(portSpec, context)
-            updateLogical(connected, true)
+            propertyChanged(connected, true)
 //        connector.open()
             //Initialize axes
             val idn = read(identity)
@@ -196,7 +196,7 @@ class PiMotionMasterDevice(
                 it.close()
             }
             port = null
-            updateLogical(connected, false)
+            propertyChanged(connected, false)
         }