diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/colour.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/colour.kt index a4f6fba..845cfd0 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/colour.kt +++ b/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(), + ) +} diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/positions.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/positions.kt index 0f80a73..2729233 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/converters/positions.kt +++ b/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, ) diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioCommonData.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioCommonData.kt index d6ffb50..f08eb79 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioCommonData.kt +++ b/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 diff --git a/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/ColourHexTest.kt b/modules/events-schema/src/commonTest/kotlin/dev/adamko/kafkatorio/events/schema/ColourHexTest.kt new file mode 100644 index 0000000..beaf512 --- /dev/null +++ b/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 { 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) + + } + + } + } + + +}) diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index 7e0bebf..0567e7f 100644 --- a/modules/events-server/build.gradle.kts +++ b/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 { "-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) +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt index 02d4520..dd9b89b 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt +++ b/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() } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt index ec6fbb8..ffad94f 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/KafkatorioTopology.kt +++ b/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) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt index 8506cac..a8b487b 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/eventProcessor.kt +++ b/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() diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt index 02be069..3de8cc6 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt +++ b/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 -> chunkTiles.chunkId.serverId } ) { tiles: ServerMapChunkTiles, colourDict: TileColourDict -> - ServerMapChunkTiles( - tiles.chunkId, - tiles.map.mapValues { (_, code) -> - colourDict.map.getOrElse(code) { - println("missing prototype $code") - ColourHex.TRANSPARENT - } + val missingProtos = mutableSetOf() + + 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 diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt index ea7459d..4987c33 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt @@ -36,6 +36,7 @@ fun factorioServerPacketStream( } } + fun splitFactorioServerPacketStream( factorioServerPacketStream: KStream ) { diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt index 3ee7439..307c915 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt +++ b/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, @@ -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(), + ) +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt index 3bae3f1..8bd855d 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt +++ b/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 ) { diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt index ac19bbd..63ce083 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt +++ b/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) { - operator fun plus(other: TileColourDict) = TileColourDict(this.map + other.map) -} +value class TileColourDict(val map: Map) -/** Aggregate [KafkatorioPacket]s into a single [FactorioServerMap] (per [serverId][FactorioServerId]). */ +/** Get the latest map-tile colours per server. */ fun tileProtoColourDictionary( factorioServerPacketStream: KStream ): KTable { // 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(), ) )