Add Pi serial port

controls-core/src/jvmMain/kotlin/space/kscience/controls/ports/ChannelPort.kt

@@ -15,7 +15,7 @@ import java.nio.channels.DatagramChannel
 import java.nio.channels.SocketChannel
 import kotlin.coroutines.CoroutineContext
 
-internal fun ByteBuffer.readArray(limit: Int = limit()): ByteArray {
+public fun ByteBuffer.toArray(limit: Int = limit()): ByteArray {
     rewind()
     val response = ByteArray(limit)
     get(response)
@@ -48,7 +48,7 @@ public class ChannelPort (
             try {
                 val num = channel.read(buffer)
                 if (num > 0) {
-                    receive(buffer.readArray(num))
+                    receive(buffer.toArray(num))
                 }
                 if (num < 0) cancel("The input channel is exhausted")
             } catch (ex: Exception) {

controls-pi/src/main/kotlin/space/kscience/controls/pi/PiPlugin.kt

@@ -0,0 +1,19 @@
+package space.kscience.controls.pi
+
+import space.kscience.dataforge.context.AbstractPlugin
+import space.kscience.dataforge.context.Context
+import space.kscience.dataforge.context.PluginFactory
+import space.kscience.dataforge.context.PluginTag
+import space.kscience.dataforge.meta.Meta
+
+public class PiPlugin : AbstractPlugin() {
+    override val tag: PluginTag get() = Companion.tag
+
+    public companion object : PluginFactory<PiPlugin> {
+
+        override val tag: PluginTag = PluginTag("controls.ports.pi", group = PluginTag.DATAFORGE_GROUP)
+
+        override fun build(context: Context, meta: Meta): PiPlugin = PiPlugin()
+
+    }
+}

controls-pi/src/main/kotlin/space/kscience/controls/pi/PiSerialPort.kt

@@ -1,14 +1,19 @@
 package space.kscience.controls.pi
 
-import com.pi4j.io.serial.FlowControl
+import com.pi4j.Pi4J
-import com.pi4j.io.serial.Parity
 import com.pi4j.io.serial.Serial
-import com.pi4j.io.serial.StopBits
+import com.pi4j.io.serial.SerialConfigBuilder
-import com.pi4j.ktx.io.open
-import com.pi4j.ktx.io.piGpioSerialProvider
 import com.pi4j.ktx.io.serial
+import kotlinx.coroutines.*
 import space.kscience.controls.ports.AbstractPort
+import space.kscience.controls.ports.Port
+import space.kscience.controls.ports.PortFactory
+import space.kscience.controls.ports.toArray
 import space.kscience.dataforge.context.Context
+import space.kscience.dataforge.context.error
+import space.kscience.dataforge.context.logger
+import space.kscience.dataforge.meta.Meta
+import java.nio.ByteBuffer
 import kotlin.coroutines.CoroutineContext
 
 public class PiSerialPort(
@@ -17,9 +22,44 @@ public class PiSerialPort(
     public val serialBuilder: () -> Serial,
 ) : AbstractPort(context, coroutineContext) {
 
-    private val serial by lazy { serialBuilder() }
+    private val serial: Serial by lazy { serialBuilder() }
+
+
+    private val listenerJob = this.scope.launch(Dispatchers.IO) {
+        val buffer = ByteBuffer.allocate(1024)
+        while (isActive) {
+            try {
+                val num = serial.read(buffer)
+                if (num > 0) {
+                    receive(buffer.toArray(num))
+                }
+                if (num < 0) cancel("The input channel is exhausted")
+            } catch (ex: Exception) {
+                logger.error(ex) { "Channel read error" }
+                delay(1000)
+            }
+        }
+    }
+
+    override suspend fun write(data: ByteArray): Unit = withContext(Dispatchers.IO) {
+        serial.write(data)
+    }
+
+    override fun close() {
+        listenerJob.cancel()
+        serial.close()
+    }
+
+    public companion object : PortFactory {
+        override val type: String get() = "pi"
+
+        public fun open(context: Context, device: String, block: SerialConfigBuilder.() -> Unit): PiSerialPort =
+            PiSerialPort(context) { Pi4J.newAutoContext().serial(device, block) }
+
+        override fun build(context: Context, meta: Meta): Port = PiSerialPort(context) {
+            Pi4J.newAutoContext().serial()
+        }
 
-    override suspend fun write(data: ByteArray) {
-        TODO()
     }
 }
+