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 5f5b68f..acb614b 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 @@ -4,7 +4,7 @@ import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream import dev.adamko.kafkatorio.processor.topology.groupMapChunks import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer -import dev.adamko.kafkatorio.processor.topology.saveMapTiles +import dev.adamko.kafkatorio.processor.topology.saveMapTiles2 import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream import java.time.Duration import java.util.Properties @@ -59,7 +59,8 @@ internal class KafkatorioTopology( private fun saveTiles() { val builder = StreamsBuilder() - val topology = saveMapTiles(builder) +// val topology = saveMapTiles(builder) + val topology = saveMapTiles2(builder) launchTopology("saveTiles", topology) } 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 3909488..ede9b1e 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 @@ -4,12 +4,17 @@ 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.schema.common.* import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kafkatorio.schema.common.ColourHex +import dev.adamko.kafkatorio.schema.common.MapChunkPosition +import dev.adamko.kafkatorio.schema.common.MapTilePosition +import dev.adamko.kafkatorio.schema.common.SurfaceIndex +import dev.adamko.kafkatorio.schema.common.toMapChunkPosition import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.streams.forEach import dev.adamko.kotka.kxs.serde +import java.awt.Color import java.awt.image.BufferedImage import java.io.File import kotlin.coroutines.CoroutineContext @@ -36,6 +41,7 @@ import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.runningFold import kotlinx.coroutines.launch import kotlinx.coroutines.supervisorScope +import kotlinx.serialization.Serializable import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.Topology import org.apache.kafka.streams.kstream.KStream @@ -51,11 +57,12 @@ private val serverMapChunkHandler = ServerMapChunkHandler() fun saveMapTiles( builder: StreamsBuilder, ): Topology { + val pid = "saveMapTiles" val groupedMapChunkTiles: KStream> = builder.stream( TOPIC_GROUPED_MAP_CHUNKS_STATE, - consumedAs("consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()) + consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()) ) groupedMapChunkTiles @@ -71,24 +78,24 @@ fun saveMapTiles( @JvmInline -private value class TilePngFilename( +value class TilePngFilename( val value: String, ) { - constructor(cmd: Cmd.CreateImage) : this(buildString { + constructor(id: ServerMapChunkId) : this(buildString { append("src/main/resources/kafkatorio-web-map") - append("/s${cmd.surfaceIndex}") - append("/z${cmd.chunkSize.zoomLevel}") - append("/x${cmd.chunkPosition.x}") - append("/y${cmd.chunkPosition.y}") + append("/s${id.surfaceIndex}") + append("/z${id.chunkSize.zoomLevel}") + append("/x${id.chunkPosition.x}") + append("/y${id.chunkPosition.y}") append(".png") }) } -private val TRANSPARENT_AWT = ColourHex.TRANSPARENT.toRgbColor().awt() +val TRANSPARENT_AWT: Color = ColourHex.TRANSPARENT.toRgbColor().awt() -private fun ColourHex.toRgbColor(): RGBColor { +fun ColourHex.toRgbColor(): RGBColor { return RGBColor( red.toInt(), green.toInt(), @@ -210,15 +217,13 @@ private class ServerMapChunkHandler : CoroutineScope { .asFlow() } .map { createCmd -> - val filename = TilePngFilename(createCmd) - val img = createMapTileImage( createCmd.chunkPosition, createCmd.tiles, createCmd.chunkSize, ) - Cmd.SaveImage(filename, img) + Cmd.SaveImage(createCmd.tilePngFilename, img) } launch { @@ -270,17 +275,31 @@ private class ServerMapChunkHandler : CoroutineScope { private sealed interface Cmd { + @Serializable data class ChunkSubdivide( val chunkId: ServerMapChunkId, val tiles: Map, ) : Cmd + @Serializable data class CreateImage( val surfaceIndex: SurfaceIndex, val chunkSize: ChunkSize, val chunkPosition: MapChunkPosition, val tiles: Map, - ) : Cmd + ) : Cmd { + val tilePngFilename: TilePngFilename by lazy { + TilePngFilename(buildString { + append("src/main/resources/kafkatorio-web-map") + append("/s${surfaceIndex}") + append("/z${chunkSize.zoomLevel}") + append("/x${chunkPosition.x}") + append("/y${chunkPosition.y}") + append(".png") + }) + } + } + data class SaveImage( val filename: TilePngFilename, diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles2.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles2.kt new file mode 100644 index 0000000..5cdd67e --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles2.kt @@ -0,0 +1,99 @@ +package dev.adamko.kafkatorio.processor.topology + +import com.sksamuel.scrimage.ImmutableImage +import com.sksamuel.scrimage.ScaleMethod +import com.sksamuel.scrimage.nio.PngWriter +import dev.adamko.kafkatorio.processor.admin.TOPIC_GROUPED_MAP_CHUNKS_STATE +import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kafkatorio.schema.common.ColourHex +import dev.adamko.kafkatorio.schema.common.toMapChunkPosition +import dev.adamko.kotka.extensions.consumedAs +import dev.adamko.kotka.extensions.streams.flatMap +import dev.adamko.kotka.extensions.streams.forEach +import dev.adamko.kotka.kxs.serde +import java.awt.image.BufferedImage +import java.io.File +import kotlin.io.path.fileSize +import kotlin.math.abs +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.Topology +import org.apache.kafka.streams.kstream.KStream + + +private const val WEB_MAP_TILE_SIZE_PX = 256 + + +fun saveMapTiles2( + builder: StreamsBuilder, +): Topology { + val pid = "saveMapTiles" + + val groupedMapChunkTiles: KStream> = + builder.stream( + TOPIC_GROUPED_MAP_CHUNKS_STATE, + consumedAs("$pid.consume.grouped-map-chunks", kxsBinary.serde(), kxsBinary.serde()) + ) + + groupedMapChunkTiles.flatMap("$pid.subdivide-chunks") { initialKey, initialChunk -> + ChunkSize.entries.flatMap { chunkSize -> + initialChunk.map.entries + .groupBy( + { (tile, _) -> tile.toMapChunkPosition(chunkSize.tilesPerChunk) } + ) { (tilePosition, colour) -> + tilePosition to colour + } + .map { (chunkPos, tiles) -> + val key = initialKey.copy( + chunkSize = chunkSize, + chunkPosition = chunkPos + ) + + val value = ServerMapChunkTiles(key, tiles.toMap()) + + key to value + } + } + }.forEach("$pid.save-tiles") { key, value -> + val filename = TilePngFilename(key) + val chunkColours = value.map + + val chunkImage = ImmutableImage.filled( + key.chunkSize.tilesPerChunk, + key.chunkSize.tilesPerChunk, + TRANSPARENT_AWT, + BufferedImage.TYPE_INT_ARGB + ) + + val chunkOriginX: Int = key.chunkPosition.x * key.chunkSize.tilesPerChunk + val chunkOriginY: Int = key.chunkPosition.y * key.chunkSize.tilesPerChunk + + chunkColours.forEach { (tilePosition, colour) -> + val rgbColour = colour.toRgbColor() + + val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX)) + val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY)) + + chunkImage.setColor(pixelX, pixelY, rgbColour) + } + + val img = chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.FastScale) + + val chunkImageFile = File(filename.value) + if (chunkImageFile.parentFile.mkdirs()) { + println("created new map tile parentFile directory ${chunkImageFile.absolutePath}") + } + + val sizeBefore = chunkImageFile.takeIf { it.exists() }?.toPath()?.fileSize() + + val savedTile = img.output(PngWriter.NoCompression, chunkImageFile) + + val sizeAfter = savedTile.takeIf { it.exists() }?.toPath()?.fileSize() + if (sizeBefore != sizeAfter) { +// println("savedTile $sizeBefore/$sizeAfter: $savedTile") + } else { + println("savedTile NO-SIZE-CHANGE $sizeBefore/$sizeAfter: $savedTile") + } + } + + return builder.build() +}