workinggggg!!!!!!!

- fix map tiles proto colours KTable
- test and tidy up Colour/ColourHex
- reset event-server application

modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/colour.kt

@@ -1,6 +1,7 @@
 package dev.adamko.kafkatorio.events.schema.converters
 
 import dev.adamko.kafkatorio.events.schema.Colour
+import dev.adamko.kafkatorio.events.schema.ColourHex
 import kotlin.math.roundToInt
 
 
@@ -15,39 +16,12 @@ fun Colour.isHexadecimal(): Boolean {
 /** True if all values are between `[0..1]`. */
 fun Colour.isPercentage(): Boolean = !isHexadecimal()
 
-fun Colour.toHexadecimal() = if (isPercentage()) {
-  Colour(
-    (red * 255).coerceIn(0f..255f),
-    (green * 255).coerceIn(0f..255f),
-    (blue * 255).coerceIn(0f..255f),
-    (alpha * 255).coerceIn(0f..255f),
-  )
-} else {
-  Colour(
-    red.coerceIn(0f..255f),
-    green.coerceIn(0f..255f),
-    blue.coerceIn(0f..255f),
-    alpha.coerceIn(0f..255f),
-  )
-}
-
-fun Colour.toPercentage() = if (isPercentage()) {
-  this
-} else {
-  Colour(
-    (red / 255).coerceIn(0f..1f),
-    (green / 255).coerceIn(0f..1f),
-    (blue / 255).coerceIn(0f..1f),
-    (alpha / 255).coerceIn(0f..1f),
-  )
-}
 
 fun Colour.toHexString(includeAlpha: Boolean = false): String {
 
-  val hexConverter: Float.() -> String = if (isHexadecimal()) {
-    Float::decimalToHex
-  } else {
-    Float::percentageToHex
+  val hexConverter: Float.() -> String = when {
+    isHexadecimal() -> Float::hexadecimalToString
+    else            -> Float::percentageToHexadecimalString
   }
 
   val length = if (includeAlpha) 9 else 7
@@ -62,8 +36,51 @@ fun Colour.toHexString(includeAlpha: Boolean = false): String {
   }
 }
 
-fun Float.decimalToHex(): String =
+fun Float.hexadecimalToString(): String =
   roundToInt().toString(16).padStart(2, ' ')
 
-fun Float.percentageToHex(): String =
+fun Float.percentageToHexadecimalString(): String =
   times(255).roundToInt().toString(16).padStart(2, ' ')
+
+
+private val ubyteMaxFloat = UByte.MAX_VALUE.toFloat()
+private val ubyteMinFloat = UByte.MIN_VALUE.toFloat()
+private val ubyteRangeFloat = ubyteMinFloat..ubyteMaxFloat
+
+
+private fun Float.toColourHex() = (this * ubyteMaxFloat)
+  .coerceIn(ubyteRangeFloat)
+  .toUInt()
+  .toUByte()
+
+
+private fun Float.coerceToColourHex() = coerceIn(ubyteRangeFloat)
+  .toUInt()
+  .toUByte()
+
+
+fun Colour.toHex(): ColourHex {
+
+  val converter: Float.() -> UByte = when {
+    isPercentage() -> Float::toColourHex
+    else           -> Float::coerceToColourHex
+  }
+
+  return ColourHex(
+    red.converter(),
+    green.converter(),
+    blue.converter(),
+    alpha.converter(),
+  )
+}
+
+private fun UByte.toPercentileFloat() = (this.toFloat() / ubyteMaxFloat).coerceIn(0f..1f)
+
+fun ColourHex.toPercentile(): Colour {
+  return Colour(
+    red.toPercentileFloat(),
+    green.toPercentileFloat(),
+    blue.toPercentileFloat(),
+    alpha.toPercentileFloat(),
+  )
+}

modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/positions.kt

@@ -1,7 +1,6 @@
 package dev.adamko.kafkatorio.events.schema.converters
 
 import dev.adamko.kafkatorio.events.schema.MAP_CHUNK_SIZE
-import dev.adamko.kafkatorio.events.schema.MapBoundingBox
 import dev.adamko.kafkatorio.events.schema.MapChunkPosition
 import dev.adamko.kafkatorio.events.schema.MapEntityPosition
 import dev.adamko.kafkatorio.events.schema.MapTilePosition
@@ -17,7 +16,7 @@ operator fun MapTilePosition.plus(addend: Int) =
   MapTilePosition(x + addend, y + addend)
 
 
-fun MapTilePosition.toMapChunkPosition(  chunkSize: Int = MAP_CHUNK_SIZE) =
+fun MapTilePosition.toMapChunkPosition(chunkSize: Int = MAP_CHUNK_SIZE) =
   MapChunkPosition(
     floor(x.toDouble() / chunkSize.toDouble()).toInt(),
     floor(y.toDouble() / chunkSize.toDouble()).toInt(),
@@ -27,7 +26,7 @@ fun MapTilePosition.toMapChunkPosition(  chunkSize: Int = MAP_CHUNK_SIZE) =
 fun MapEntityPosition.toMapTilePosition() =
   MapChunkPosition(floor(x).toInt(), floor(y).toInt())
 
-fun MapEntityPosition.toMapChunkPosition( chunkSize: Int = MAP_CHUNK_SIZE) =
+fun MapEntityPosition.toMapChunkPosition(chunkSize: Int = MAP_CHUNK_SIZE) =
   MapChunkPosition(
     floor(x / chunkSize.toDouble()).toInt(),
     floor(y / chunkSize.toDouble()).toInt(),
@@ -44,8 +43,8 @@ operator fun MapChunkPosition.minus(subtrahend: Int) =
   MapChunkPosition(x - subtrahend, y - subtrahend)
 
 
-fun MapChunkPosition.leftTopTile( chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition =
-    MapTilePosition(
+fun MapChunkPosition.leftTopTile(chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition =
+  MapTilePosition(
     x * chunkSize,
     y * chunkSize,
   )

modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioCommonData.kt

@@ -1,6 +1,5 @@
 package dev.adamko.kafkatorio.events.schema
 
-import dev.adamko.kafkatorio.events.schema.converters.toHexadecimal
 import kotlinx.serialization.EncodeDefault
 import kotlinx.serialization.Serializable
 
@@ -69,20 +68,10 @@ data class Colour(
   val blue: Float = 0f,
   @EncodeDefault
   val alpha: Float = 1f,
-) {
-
-  fun toHex() = toHexadecimal().let {
-    ColourHex(
-      it.red.toUInt().toUByte(),
-      it.green.toUInt().toUByte(),
-      it.blue.toUInt().toUByte(),
-      it.alpha.toUInt().toUByte(),
-    )
-  }
-}
+)
 
 
-/** Size-efficient version of [Colour] */
+/** Size-efficient version of [Colour] (`4*4` bytes vs `4*1` bytes) */
 @Serializable
 data class ColourHex(
   @EncodeDefault

modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/ColourHexTest.kt

@@ -0,0 +1,86 @@
+package dev.adamko.kafkatorio.events.schema
+
+import dev.adamko.kafkatorio.events.schema.converters.toHex
+import dev.adamko.kafkatorio.events.schema.converters.toPercentile
+import io.kotest.assertions.withClue
+import io.kotest.core.spec.style.BehaviorSpec
+import io.kotest.matchers.floats.shouldBeWithinPercentageOf
+import io.kotest.matchers.should
+import io.kotest.matchers.shouldBe
+import io.kotest.property.checkAll
+import kotlinx.serialization.decodeFromString
+
+class ColourHexTest : BehaviorSpec({
+
+  Given("colour json") {
+
+    //Language=JSON
+    val colourJson = """
+      {
+        "red": 0.1,
+        "green": 0.4,
+        "blue": 0.7,
+        "alpha": 0.6
+      }
+    """.trimIndent()
+
+    val colour: Colour = jsonMapperKafkatorio.decodeFromString(colourJson)
+
+    Then("expect colour is decoded") {
+      colour shouldBe Colour(
+        0.1f,
+        0.4f,
+        0.7f,
+        0.6f,
+      )
+    }
+
+    Then("expect Colour can be converted to ColourHex and back") {
+      val colourHex = colour.toHex()
+      colourHex shouldBe ColourHex(
+        25u,
+        102u,
+        178u,
+        153u,
+      )
+
+      colourHex.toPercentile() should { back ->
+        withClue(back) {
+
+          val tolerance = 2.6 // 2.6%, because 255/100 = 2.55
+
+          withClue("red") { back.red.shouldBeWithinPercentageOf(0.1f, tolerance) }
+          withClue("green") { back.green.shouldBeWithinPercentageOf(0.4f, tolerance) }
+          withClue("blue") { back.blue.shouldBeWithinPercentageOf(0.7f, tolerance) }
+          withClue("alpha") { back.alpha.shouldBeWithinPercentageOf(0.6f, tolerance) }
+        }
+      }
+    }
+  }
+
+  Given("any colour json") {
+    Then("expect can always convert") {
+
+      checkAll<UByte, UByte, UByte, UByte> { r, g, b, a ->
+
+        //Language=JSON
+        val colourJson = """
+          {
+            "red":   ${r.toFloat() / UByte.MAX_VALUE.toFloat()},
+            "green": ${g.toFloat() / UByte.MAX_VALUE.toFloat()},
+            "blue":  ${b.toFloat() / UByte.MAX_VALUE.toFloat()},
+            "alpha": ${a.toFloat() / UByte.MAX_VALUE.toFloat()}
+          }
+        """.trimIndent()
+
+        val colour: Colour = jsonMapperKafkatorio.decodeFromString(colourJson)
+        val colourHex = colour.toHex()
+        colourHex shouldBe ColourHex(r, g, b, a)
+
+      }
+
+    }
+  }
+
+
+})

modules/events-server/build.gradle.kts

@@ -1,4 +1,5 @@
 import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
+import org.jetbrains.kotlin.util.parseSpaceSeparatedArgs
 
 plugins {
   dev.adamko.kafkatorio.lang.`kotlin-jvm`
@@ -64,3 +65,19 @@ tasks.withType<KotlinCompile> {
     "-opt-in=kotlinx.coroutines.FlowPreview",
   )
 }
+
+val kafkatorioEventsServerKafkaForceReset by tasks.registering(Exec::class) {
+  group = project.name
+
+  val cmd = """
+    /kafka/bin/kafka-streams-application-reset.sh --application-id kafkatorio-events-processor --input-topics factorio-server-log --force
+  """.trimIndent()
+
+  logging.captureStandardOutput(LogLevel.LIFECYCLE)
+
+  commandLine = parseSpaceSeparatedArgs(""" docker exec -d kafka $cmd """)
+}
+
+tasks.run.configure {
+  dependsOn(kafkatorioEventsServerKafkaForceReset)
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt

@@ -12,11 +12,11 @@ import org.apache.kafka.clients.admin.TopicListing
 
 private val logger = KotlinLogging.logger {}
 
-fun main() {
-  CreateTopics.create()
-}
+//fun main() {
+//  KafkaAdmin.createKafkatorioTopics()
+//}
 
-object CreateTopics {
+object KafkaAdmin {
 
   private val kafkaAdmin: Admin =
     Admin.create(
@@ -26,7 +26,7 @@ object CreateTopics {
       )
     )
 
-  fun create() {
+  fun createKafkatorioTopics() {
 
     val currentTopics = currentTopics().map { it.name() }
 

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt

@@ -16,6 +16,7 @@ class KafkatorioTopology(
   private val builder: StreamsBuilder = StreamsBuilder(),
 ) {
 
+
   companion object {
     const val sourceTopic = "factorio-server-log"
   }
@@ -23,11 +24,8 @@ class KafkatorioTopology(
 
   fun build() {
 
-
     val packets = factorioServerPacketStream(builder)
 
-
-
     splitFactorioServerPacketStream(packets)
 
     playerUpdatesToWsServer(websocketServer, builder)

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt

@@ -1,7 +1,7 @@
 package dev.adamko.kafkatorio.processor
 
 
-import dev.adamko.kafkatorio.admin.CreateTopics
+import dev.adamko.kafkatorio.admin.KafkaAdmin
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
 import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer
 import kotlinx.coroutines.runBlocking
@@ -9,11 +9,13 @@ import org.http4k.server.PolyHandler
 import org.http4k.server.Undertow
 import org.http4k.server.asServer
 
+
 val appProps = ApplicationProperties()
 
+
 fun main(): Unit = runBlocking {
 
-  CreateTopics.create()
+  KafkaAdmin.createKafkatorioTopics()
 
   val wsServer = WebsocketServer()
   val tileServer = WebMapTileServer()

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt

@@ -76,11 +76,6 @@ fun groupTilesIntoChunksWithColours(
 
         val zoomLevel = ZoomLevel.ZOOM_0
 
-//
-//        ZoomLevel
-//          .values
-//          .mapNotNull { zoomLevel ->
-
         mapTileData
           ?.tiles
           ?.groupBy { tile ->
@@ -101,7 +96,7 @@ fun groupTilesIntoChunksWithColours(
 
             chunkId to chunkTiles
           } ?: listOf()
-//          }.flatten()
+
       }
       .groupByKey(
         groupedAs("server-map-data.tiles.group", kxsBinary.serde(), kxsBinary.serde())
@@ -121,15 +116,20 @@ fun groupTilesIntoChunksWithColours(
         foreignKeyExtractor = { chunkTiles: ServerMapChunkTiles<TileProtoHashCode> -> chunkTiles.chunkId.serverId }
       ) { tiles: ServerMapChunkTiles<TileProtoHashCode>, colourDict: TileColourDict ->
 
-        ServerMapChunkTiles(
-          tiles.chunkId,
-          tiles.map.mapValues { (_, code) ->
-            colourDict.map.getOrElse(code) {
-              println("missing prototype $code")
-              ColourHex.TRANSPARENT
-            }
+        val missingProtos = mutableSetOf<TileProtoHashCode>()
+
+        val tileColours = tiles.map.mapValues { (_, code) ->
+          colourDict.map.getOrElse(code) {
+            missingProtos.add(code)
+            ColourHex.TRANSPARENT
           }
-        )
+        }
+
+        if (missingProtos.isNotEmpty()) {
+          println("missing ${missingProtos.size} tile prototypes: ${missingProtos.joinToString { "${it.code}" }}")
+        }
+
+        ServerMapChunkTiles(tiles.chunkId, tileColours)
       }
 
   return colourisedChunkTable

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt

@@ -36,6 +36,7 @@ fun factorioServerPacketStream(
     }
 }
 
+
 fun splitFactorioServerPacketStream(
   factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>
 ) {

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt

@@ -2,6 +2,7 @@ package dev.adamko.kafkatorio.processor.topology
 
 import com.sksamuel.scrimage.ImmutableImage
 import com.sksamuel.scrimage.ScaleMethod
+import com.sksamuel.scrimage.color.RGBColor
 import com.sksamuel.scrimage.nio.PngWriter
 import dev.adamko.kafkatorio.events.schema.ColourHex
 import dev.adamko.kafkatorio.events.schema.MapChunkPosition
@@ -20,7 +21,6 @@ import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.conflate
 import kotlinx.coroutines.flow.debounce
 import kotlinx.coroutines.flow.distinctUntilChanged
-import kotlinx.coroutines.flow.distinctUntilChangedBy
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
@@ -63,7 +63,7 @@ fun saveMapTiles(
       chunkedTiles
         .conflate()
         .distinctUntilChanged()
-        .debounce(5.seconds)
+        .debounce(1.seconds)
         .map {
           println("createMapTileImage ${it.chunkId}")
           createMapTileImage(it)
@@ -164,6 +164,9 @@ private fun createMapTileImage(
 }
 
 
+private val TRANSPARENT_AWT = ColourHex.TRANSPARENT.toRgbColor().awt()
+
+
 private fun createMapTileImage(
   chunkPosition: MapChunkPosition,
   chunkColours: Map<MapTilePosition, ColourHex>,
@@ -173,7 +176,7 @@ private fun createMapTileImage(
   val chunkImage = ImmutableImage.filled(
     z.chunkSize,
     z.chunkSize,
-    ColourHex.TRANSPARENT.toRgbColor().awt(),
+    TRANSPARENT_AWT,
     BufferedImage.TYPE_INT_ARGB
   )
 
@@ -196,3 +199,13 @@ private fun createMapTileImage(
 
   return chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.Progressive)
 }
+
+
+private fun ColourHex.toRgbColor(): RGBColor {
+  return RGBColor(
+    red.toInt(),
+    green.toInt(),
+    blue.toInt(),
+    alpha.toInt(),
+  )
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt

@@ -1,8 +1,5 @@
 package dev.adamko.kafkatorio.processor.topology
 
-import com.sksamuel.scrimage.color.RGBColor
-import dev.adamko.kafkatorio.events.schema.Colour
-import dev.adamko.kafkatorio.events.schema.ColourHex
 import kotlin.math.pow
 import kotlin.math.roundToInt
 import kotlinx.serialization.Serializable
@@ -24,16 +21,6 @@ value class FactorioServerId(val id: String) {
 }
 
 
-fun ColourHex.toRgbColor(): RGBColor {
-  return RGBColor(
-    red.toInt(),
-    green.toInt(),
-    blue.toInt(),
-    alpha.toInt(),
-  )
-}
-
-
 enum class ZoomLevel(
   val level: Int
 ) {

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt

@@ -5,8 +5,10 @@ import dev.adamko.kafkatorio.events.schema.FactorioPrototypes
 import dev.adamko.kafkatorio.events.schema.KafkatorioPacket
 import dev.adamko.kafkatorio.events.schema.MapTile
 import dev.adamko.kafkatorio.events.schema.MapTilePrototype
+import dev.adamko.kafkatorio.events.schema.converters.toHex
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kotka.extensions.materializedAs
+import dev.adamko.kotka.extensions.streams.filter
 import dev.adamko.kotka.extensions.streams.mapValues
 import dev.adamko.kotka.extensions.streams.toTable
 import dev.adamko.kotka.kxs.serde
@@ -22,9 +24,10 @@ value class PrototypeName(val name: String) {
 }
 
 
+/** Use the hashcode of Prototypes names as keys - more efficient storage */
 @JvmInline
 @Serializable
-value class TileProtoHashCode(val code: Int) {
+value class TileProtoHashCode private constructor(val code: Int) {
   constructor(prototype: MapTilePrototype) : this(prototype.name.hashCode())
   constructor(tile: MapTile) : this(tile.prototypeName.hashCode())
 }
@@ -32,19 +35,17 @@ value class TileProtoHashCode(val code: Int) {
 
 @JvmInline
 @Serializable
-value class TileColourDict(val map: Map<TileProtoHashCode, ColourHex>) {
-  operator fun plus(other: TileColourDict) = TileColourDict(this.map + other.map)
-}
+value class TileColourDict(val map: Map<TileProtoHashCode, ColourHex>)
 
 
-/** Aggregate [KafkatorioPacket]s into a single [FactorioServerMap] (per [serverId][FactorioServerId]). */
+/** Get the latest map-tile colours per server. */
 fun tileProtoColourDictionary(
   factorioServerPacketStream: KStream<FactorioServerId, KafkatorioPacket>
 ): KTable<FactorioServerId, TileColourDict> {
 
   // get all the prototypes
   return factorioServerPacketStream
-    .mapValues("server-map-data.tile-prototypes.mapValues") { _, packet ->
+    .mapValues("server-map-data.tile-prototypes.mapValues") { _, packet: KafkatorioPacket ->
       val map = when (packet) {
         is FactorioPrototypes -> packet
           .prototypes
@@ -53,13 +54,17 @@ fun tileProtoColourDictionary(
         else                  -> mapOf()
       }
       TileColourDict(map)
+    }.filter("server-map-data.tile-prototypes.filterMapTileProtos") { _, dict: TileColourDict ->
+      dict.map.isNotEmpty()
+    }.peek { serverId, dict: TileColourDict ->
+      println("server $serverId has TileColourDict[${dict.map.size}]: ${dict.map.entries.joinToString()}")
     }
     .toTable(
       "server-map-data.tile-prototypes",
       materializedAs(
         "server-map-data.tile-prototypes.store",
         jsonMapper.serde(),
-        jsonMapper.serde()
+        jsonMapper.serde(),
       )
     )