diff --git a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/object-data.d.ts b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/object-data.d.ts index 4d5583a..4585cfa 100644 --- a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/object-data.d.ts +++ b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/object-data.d.ts @@ -1,6 +1,6 @@ // Generated by TypeScriptGenerator - do not edit this file manually -type ObjectName = "LuaPlayer" | "LuaEntity" | "LuaSurface" | "MapChunk" | "LuaTile" | "LuaTiles" | "ConsoleChatMessage"; +type ObjectName = "LuaPlayer" | "LuaEntity" | "LuaSurface" | "MapChunk" | "LuaTiles" | "ConsoleChatMessage"; interface FactorioObjectData { objectName: ObjectName; @@ -37,12 +37,6 @@ interface SurfaceData extends FactorioObjectData { objectName: ObjectName; } -interface MapTile extends FactorioObjectData { - objectName: ObjectName; - position: MapTilePosition; - prototypeName: string; -} - interface MapTiles extends FactorioObjectData { objectName: ObjectName; surfaceIndex: int; diff --git a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/schema.d.ts b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/schema.d.ts index ad34892..da32cea 100644 --- a/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/schema.d.ts +++ b/modules/events-mod/src/main/typescript/generated/kafkatorio-schema/schema.d.ts @@ -41,3 +41,9 @@ interface MapChunkPosition { x: int; y: int; } + +interface MapTile { + objectName: string; + position: MapTilePosition; + prototypeName: string; +} diff --git a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioLuaObjects.kt b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioLuaObjects.kt index d142fbb..ac89d09 100644 --- a/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioLuaObjects.kt +++ b/modules/events-schema/src/commonMain/kotlin/dev/adamko/kafkatorio/events/schema/factorioLuaObjects.kt @@ -38,7 +38,7 @@ sealed class FactorioObjectData { LuaEntity, LuaSurface, MapChunk, - LuaTile, + // LuaTile, LuaTiles, ConsoleChatMessage, } @@ -66,7 +66,7 @@ object FactorioObjectDataJsonSerializer : JsonContentPolymorphicSerializer EntityData.serializer() FactorioObjectData.ObjectName.LuaSurface -> SurfaceData.serializer() FactorioObjectData.ObjectName.MapChunk -> MapChunk.serializer() - FactorioObjectData.ObjectName.LuaTile -> MapTile.serializer() +// FactorioObjectData.ObjectName.LuaTile -> MapTile.serializer() FactorioObjectData.ObjectName.LuaTiles -> MapTiles.serializer() FactorioObjectData.ObjectName.ConsoleChatMessage -> ConsoleChatMessage.serializer() } @@ -141,9 +141,11 @@ data class MapTiles( data class MapTile( val prototypeName: String, val position: MapTilePosition, -) : FactorioObjectData() { - @EncodeDefault - override val objectName = ObjectName.LuaTile +) +// : FactorioObjectData() +{ + // @EncodeDefault + val objectName: String = "" } @Serializable diff --git a/modules/events-server/build.gradle.kts b/modules/events-server/build.gradle.kts index 64c0fcf..7e0bebf 100644 --- a/modules/events-server/build.gradle.kts +++ b/modules/events-server/build.gradle.kts @@ -17,7 +17,7 @@ val projectId: String by project.extra @Suppress("UnstableApiUsage") // platform + version-catalog is incubating dependencies { - implementation("org.apache.kafka:kafka-streams:3.0.0") + implementation("org.apache.kafka:kafka-streams:3.1.0") implementation(projects.modules.eventsSchema) @@ -38,8 +38,8 @@ dependencies { implementation("com.sksamuel.scrimage:scrimage-core:4.0.24") -// val kotkaVersion = "0.0.11" - val kotkaVersion = "store-iterators-SNAPSHOT" + val kotkaVersion = "v0.0.12" +// val kotkaVersion = "store-iterators-SNAPSHOT" implementation("com.github.adamko-dev.kotka-streams:kotka-streams:$kotkaVersion") { isChanging = true } 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 89e9438..9049e61 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 @@ -1,22 +1,14 @@ package dev.adamko.kafkatorio.processor -import dev.adamko.kafkatorio.events.schema.Colour -import dev.adamko.kafkatorio.events.schema.MapTile -import dev.adamko.kafkatorio.processor.topology.PrototypeName -import dev.adamko.kafkatorio.processor.topology.TileUpdateRecordKey -import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPixels -import dev.adamko.kafkatorio.processor.topology.WebMapTileChunkPosition -import dev.adamko.kafkatorio.processor.topology.aggregateWebMapTiles -import dev.adamko.kafkatorio.processor.topology.allMapTilesTable +import dev.adamko.kafkatorio.processor.topology.buildFactorioServerMap +import dev.adamko.kafkatorio.processor.topology.factorioServerPacketStream import dev.adamko.kafkatorio.processor.topology.playerUpdatesToWsServer -import dev.adamko.kafkatorio.processor.topology.saveTileImages -import dev.adamko.kafkatorio.processor.topology.splitFactorioServerLog -import dev.adamko.kafkatorio.processor.topology.tilePrototypeColourTable +import dev.adamko.kafkatorio.processor.topology.saveMapTiles +import dev.adamko.kafkatorio.processor.topology.splitFactorioServerPacketStream import java.time.Duration import org.apache.kafka.streams.KafkaStreams import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.TopologyDescription -import org.apache.kafka.streams.kstream.KTable class KafkatorioTopology( @@ -30,17 +22,25 @@ class KafkatorioTopology( fun build() { - splitFactorioServerLog(builder) + val packets = factorioServerPacketStream(builder) + + splitFactorioServerPacketStream(packets) playerUpdatesToWsServer(websocketServer, builder) - val allMapTilesTable: KTable = allMapTilesTable(builder) - val tilePrototypeColourTable: KTable = tilePrototypeColourTable(builder) + val serverMapDataTable = buildFactorioServerMap(packets) - val webMapTiles: KTable = - aggregateWebMapTiles(allMapTilesTable, tilePrototypeColourTable) + saveMapTiles(serverMapDataTable) - saveTileImages(webMapTiles) + + // the old way... +// val allMapTilesTable: KTable = allMapTilesTable(builder) +// val tilePrototypeColourTable: KTable = tilePrototypeColourTable(builder) +// +// val webMapTiles: KTable = +// aggregateWebMapTiles(allMapTilesTable, tilePrototypeColourTable) +// +// saveTileImages(webMapTiles) val topology = builder.build() 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 new file mode 100644 index 0000000..7e9df42 --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioGameState.kt @@ -0,0 +1,127 @@ +package dev.adamko.kafkatorio.processor.topology + +import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate +import dev.adamko.kafkatorio.events.schema.FactorioEvent +import dev.adamko.kafkatorio.events.schema.FactorioPrototypes +import dev.adamko.kafkatorio.events.schema.KafkatorioPacket +import dev.adamko.kafkatorio.events.schema.MapChunk +import dev.adamko.kafkatorio.events.schema.MapTilePosition +import dev.adamko.kafkatorio.events.schema.MapTilePrototype +import dev.adamko.kafkatorio.events.schema.MapTiles +import dev.adamko.kafkatorio.processor.serdes.jsonMapper +import dev.adamko.kafkatorio.processor.serdes.kxsBinary +import dev.adamko.kotka.extensions.groupedAs +import dev.adamko.kotka.extensions.materializedAs +import dev.adamko.kotka.extensions.materializedWith +import dev.adamko.kotka.extensions.streams.aggregate +import dev.adamko.kotka.extensions.tables.mapValues +import dev.adamko.kotka.kxs.serde +import kotlinx.serialization.Serializable +import org.apache.kafka.streams.kstream.KStream +import org.apache.kafka.streams.kstream.KTable + + +@Serializable +data class FactorioServerMap( + val serverId: FactorioServerId, + val tilePrototypes: Map, + val surfaces: Map, +) + + +@Serializable +data class FactorioMapSurface( + val index: SurfaceIndex, + val tiles: Map, +) + +/** Aggregate [KafkatorioPacket]s into a single [FactorioServerMap] (per [serverId][FactorioServerId]). */ +fun buildFactorioServerMap( + factorioServerPacketStream: KStream +): KTable { + + return factorioServerPacketStream + .groupByKey( + groupedAs("factorio-server-map-data-grouping", jsonMapper.serde(), jsonMapper.serde()) + ) + .aggregate( + "factorio-server-map-data-aggregate", + materializedAs("factorio-server-map-aggregate-store", kxsBinary.serde(), kxsBinary.serde()), + { FactorioServerMapAggregator() }, + ) { _: FactorioServerId, packet: KafkatorioPacket, aggregate: FactorioServerMapAggregator -> + aggregate + packet + } + .mapValues( + "finalise-factorio-server-map-aggregate", + materializedWith(kxsBinary.serde(), kxsBinary.serde()) + ) { readOnlyKey, aggregate -> + + val surfaces = aggregate.surfaces.mapValues { (_, surface) -> + FactorioMapSurface(surface.index, surface.tiles) + } + + FactorioServerMap(readOnlyKey, aggregate.tilePrototypes, surfaces) + } + +} + + +@Serializable +private data class FactorioServerMapAggregator( + val tilePrototypes: MutableMap = mutableMapOf(), + val surfaces: MutableMap = mutableMapOf(), +) { + + operator fun plus(packet: KafkatorioPacket): FactorioServerMapAggregator { + return when (packet) { + is FactorioEvent -> { + when (val data = packet.data) { + is MapChunk -> plus(data) + is MapTiles -> plus(data) +// is MapTile -> {} // hmmm... no surface index + else -> this + } + } + is FactorioPrototypes -> plus(packet) + is FactorioConfigurationUpdate -> this + } + } + + operator fun plus(mapTiles: MapTiles): FactorioServerMapAggregator { +// println("FactorioServerMapAggregator: adding ${mapTiles.tiles.size} MapTiles from surface ${mapTiles.surfaceIndex}") + + val surfaceIndex = SurfaceIndex(mapTiles.surfaceIndex) + + val surface = surfaces.getOrPut(surfaceIndex) { FactorioMapSurfaceAggregator(surfaceIndex) } + + val surfaceTiles = mapTiles.tiles.associate { it.position to PrototypeName(it.prototypeName) } + surface.tiles.putAll(surfaceTiles) + + return this + } + + operator fun plus(mapChunk: MapChunk): FactorioServerMapAggregator { +// println("FactorioServerMapAggregator: adding mapChunk ${mapChunk.position}") + return plus(mapChunk.tiles) + } + + operator fun plus(packet: FactorioPrototypes): FactorioServerMapAggregator { +// println("FactorioServerMapAggregator: adding FactorioPrototypes") + + val packetPrototypes = packet.prototypes + .filterIsInstance() + .associateBy { PrototypeName(it.name) } + + tilePrototypes.putAll(packetPrototypes) + + return this + } + +} + + +@Serializable +private data class FactorioMapSurfaceAggregator( + val index: SurfaceIndex, + val tiles: MutableMap = mutableMapOf(), +) diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt index 6a6ee07..97242dc 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt @@ -35,24 +35,24 @@ fun allMapTilesTable( builder: StreamsBuilder ): KTable { - val luaTilesUpdatesStream: KStream = - builder.stream( + val luaTilesUpdatesStream: KStream = + builder.stream( "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", consumedAs("consume-map-tiles-packets", jsonMapper.serde(), jsonMapper.serde()) ) - .filter("events.filter.map-tiles") { _: FactorioPacketKey, event: FactorioEvent -> + .filter("events.filter.map-tiles") { _: FactorioServerId, event: FactorioEvent -> event.data is MapTiles } .mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! } - val chunkTilesUpdateStream: KStream = - builder.stream( + val chunkTilesUpdateStream: KStream = + builder.stream( "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", consumedAs("consume-map-chunk-packets", jsonMapper.serde(), jsonMapper.serde()) ) - .filter("events.filter.map-chunks") { _: FactorioPacketKey, event: FactorioEvent -> + .filter("events.filter.map-chunks") { _: FactorioServerId, event: FactorioEvent -> event.data is MapChunk } .mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> 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 229c3f4..ea7459d 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 @@ -1,17 +1,9 @@ package dev.adamko.kafkatorio.processor.topology -import dev.adamko.kafkatorio.events.schema.ConsoleChatMessage -import dev.adamko.kafkatorio.events.schema.EntityData import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate import dev.adamko.kafkatorio.events.schema.FactorioEvent import dev.adamko.kafkatorio.events.schema.FactorioPrototypes import dev.adamko.kafkatorio.events.schema.KafkatorioPacket -import dev.adamko.kafkatorio.events.schema.MapChunk -import dev.adamko.kafkatorio.events.schema.MapTile -import dev.adamko.kafkatorio.events.schema.MapTiles -import dev.adamko.kafkatorio.events.schema.PlayerData -import dev.adamko.kafkatorio.events.schema.SurfaceData -import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition import dev.adamko.kafkatorio.processor.KafkatorioTopology import dev.adamko.kafkatorio.processor.serdes.jsonMapper import dev.adamko.kotka.extensions.consumedAs @@ -19,25 +11,18 @@ import dev.adamko.kotka.extensions.producedAs import dev.adamko.kotka.extensions.streams.map import dev.adamko.kotka.extensions.streams.to import dev.adamko.kotka.kxs.serde -import kotlinx.serialization.Serializable import kotlinx.serialization.decodeFromString import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.kstream.KStream import org.apache.kafka.streams.processor.RecordContext -@Serializable -data class FactorioPacketKey( - val serverId: FactorioServerId, - /** arbitrary distinctive key */ - val key: String, -) - -fun splitFactorioServerLog( +fun factorioServerPacketStream( builder: StreamsBuilder = StreamsBuilder(), -) { +): KStream { - builder.stream( + return builder.stream( KafkatorioTopology.sourceTopic, consumedAs( "read-raw-packets-from-server", @@ -46,18 +31,22 @@ fun splitFactorioServerLog( ) ) .map("decode-packets") { serverId: String, value: String -> -// println("Mapping $readOnlyKey:$value") val packet = jsonMapper.decodeFromString(value) - val key: String = createPacketKey(packet) - FactorioPacketKey(FactorioServerId(serverId), key) to packet + FactorioServerId(serverId) to packet } +} + +fun splitFactorioServerPacketStream( + factorioServerPacketStream: KStream +) { + factorioServerPacketStream .to( producedAs( "split-server-log", jsonMapper.serde(), jsonMapper.serde(), ) - ) { _: FactorioPacketKey, value: KafkatorioPacket, _: RecordContext -> + ) { _: FactorioServerId, value: KafkatorioPacket, _: RecordContext -> // println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}") when (value) { is FactorioEvent -> @@ -68,28 +57,34 @@ fun splitFactorioServerLog( "kafkatorio.${value.packetType.name}.all" } } - } -private fun createPacketKey(packet: KafkatorioPacket): String { - - return when (packet) { - is FactorioEvent -> factorioEventKey(packet) - is FactorioConfigurationUpdate, - is FactorioPrototypes -> packet.hashCode().toString() - } -} - -private fun factorioEventKey(event: FactorioEvent): String { - val id: String = when (val data = event.data) { - is ConsoleChatMessage -> data.authorPlayerIndex?.toString() - is EntityData -> data.type - is MapChunk -> data.position.toString() - is MapTile -> data.position.toString() - is MapTiles -> data.tiles.firstOrNull()?.position?.toMapChunkPosition()?.toString() - is PlayerData -> (data.characterUnitNumber?.toString() ?: data.name) - is SurfaceData -> data.index.toString() - } ?: event.tick.toString() - - return id -} +//@Serializable +//data class FactorioPacketKey( +// val serverId: FactorioServerId, +// /** arbitrary distinctive key */ +// val key: String, +//) +// +//private fun createPacketKey(packet: KafkatorioPacket): String { +// +// return when (packet) { +// is FactorioEvent -> factorioEventKey(packet) +// is FactorioConfigurationUpdate, +// is FactorioPrototypes -> packet.hashCode().toString() +// } +//} +// +//private fun factorioEventKey(event: FactorioEvent): String { +// val id: String = when (val data = event.data) { +// is ConsoleChatMessage -> data.authorPlayerIndex?.toString() +// is EntityData -> data.type +// is MapChunk -> data.position.toString() +//// is MapTile -> data.position.toString() +// is MapTiles -> data.tiles.firstOrNull()?.position?.toMapChunkPosition()?.toString() +// is PlayerData -> (data.characterUnitNumber?.toString() ?: data.name) +// is SurfaceData -> data.index.toString() +// } ?: event.tick.toString() +// +// return id +//} 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 new file mode 100644 index 0000000..2aabdf4 --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/flowMapTiles.kt @@ -0,0 +1,198 @@ +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.events.schema.Colour +import dev.adamko.kafkatorio.events.schema.MapChunkPosition +import dev.adamko.kafkatorio.events.schema.MapTilePosition +import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition +import java.awt.image.BufferedImage +import java.io.File +import kotlin.math.abs +import kotlin.time.Duration.Companion.seconds +import kotlinx.coroutines.CoroutineName +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.asFlow +import kotlinx.coroutines.flow.conflate +import kotlinx.coroutines.flow.filterNotNull +import kotlinx.coroutines.flow.flatMapMerge +import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.mapLatest +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import org.apache.kafka.streams.kstream.KTable + + +private const val WEB_MAP_TILE_SIZE_PX = 256 + +private val cScope = CoroutineScope( + Dispatchers.Default + SupervisorJob() + CoroutineName("f-server-map-updates") +) + +/** A bridge between Kafka and Kotlin */ +private val serverMapDataFlow: MutableStateFlow = MutableStateFlow(null) + + +fun saveMapTiles( + serverMapTable: KTable +) { + + serverMapTable + .toStream() + .foreach { _, value -> + runBlocking(Dispatchers.Default) { + serverMapDataFlow.emit(value) + } + } + + cScope.launch { + println("launching serverMapDataFlow merging") + + + serverMapDataFlow + .filterNotNull() + + serverMapDataFlow + .filterNotNull() + .conflate() +// .transform { serverMapData -> + .flatMapMerge(10) { serverMapData -> + println("flatMapMerge serverMapData ${serverMapData.serverId}, surfaces ${serverMapData.surfaces.keys}") + + val prototypes = serverMapData.tilePrototypes + + serverMapData + .surfaces + .values +// .forEach { surfaceData -> + .asFlow() + .flatMapMerge(10) { surfaceData -> + println("flatMapMerge surfaceData ${surfaceData.index}") + + // get all tile colours for the surface + val surfaceTileColours = surfaceData.tiles + .entries + .mapNotNull { (tilePosition, prototypeName) -> + prototypes[prototypeName]?.let { prototype -> + tilePosition to prototype.mapColour + } + } + println("surfaceTileColours.size ${surfaceTileColours.size}") + + // split a surface into chunks, per zoom-level + ZoomLevel.values +// .forEach { zoom -> + .asFlow() + .flatMapMerge(10) { zoom -> + println("flatMapMerge zoom $zoom") + + surfaceTileColours + .groupBy { (tilePosition, _) -> + tilePosition.toMapChunkPosition(zoom.chunkSize) + } + .entries +// .forEach { (chunkPosition, tileColours) -> + .asFlow() + .mapLatest { (chunkPosition, tileColours) -> + println("mapLatest chunkPosition $chunkPosition") + + // mapLatest + delay = an alternative to debounce-per-key? + delay(30.seconds) + + println("emitting ChunkData $chunkPosition") + + ChunkData( + surfaceData.index, + zoom, + chunkPosition, + tileColours.toMap() + ) + } + } + } + } + .onEach { chunkData -> + + println("starting to save chunkData ${chunkData.chunkPosition}...") + + val chunkImageFile = File(chunkData.filename.value) + if (chunkImageFile.parentFile.mkdirs()) { + println("created new map tile parentFile directory ${chunkImageFile.absolutePath}") + } + + val image = + createMapTileImage(chunkData.chunkPosition, chunkData.tileColours, chunkData.zoomLevel) + + println("saving map tile $chunkImageFile") + val savedTile = image.output(PngWriter.NoCompression, chunkImageFile) + println("savedTile: $savedTile") + + } + .launchIn(this) + } +} + + +@JvmInline +private value class TilePngFilename( + val value: String, +) { + + constructor( + surfaceIndex: SurfaceIndex, + zoom: Int, + chunkX: Int, + chunkY: Int, + ) : this("src/main/resources/kafkatorio-web-map/s${surfaceIndex.index}/z$zoom/x${chunkX}/y${chunkY}.png") + +} + + +private data class ChunkData( + val surfaceIndex: SurfaceIndex, + val zoomLevel: ZoomLevel, + val chunkPosition: MapChunkPosition, + val tileColours: Map, +) { + val filename = TilePngFilename(surfaceIndex, zoomLevel.level, chunkPosition.x, chunkPosition.y) +} + +private fun createMapTileImage( + chunkPosition: MapChunkPosition, + chunkColours: Map, + z: ZoomLevel +): ImmutableImage { + + + val chunkImage = ImmutableImage.filled( + z.chunkSize, + z.chunkSize, + COLOUR_TRANSPARENT.awt(), + BufferedImage.TYPE_INT_ARGB + ) + + val chunkOriginX: Int = chunkPosition.x * z.chunkSize + val chunkOriginY: Int = chunkPosition.y * z.chunkSize + + 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 + ) + } + + return chunkImage.scaleTo(WEB_MAP_TILE_SIZE_PX, WEB_MAP_TILE_SIZE_PX, ScaleMethod.Progressive) +} 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 fb785c6..4f67bc5 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 @@ -2,9 +2,11 @@ package dev.adamko.kafkatorio.processor.topology import com.sksamuel.scrimage.color.RGBColor import dev.adamko.kafkatorio.events.schema.Colour +import kotlin.math.pow import kotlin.math.roundToInt import kotlinx.serialization.Serializable + @Serializable @JvmInline value class SurfaceIndex( @@ -13,12 +15,14 @@ value class SurfaceIndex( override fun toString() = "$index" } + @JvmInline @Serializable value class FactorioServerId(val id: String) { override fun toString() = id } + fun Colour.toRgbColor(): RGBColor { return RGBColor( red.roundToInt(), @@ -27,3 +31,26 @@ fun Colour.toRgbColor(): RGBColor { alpha.roundToInt(), ) } + + +enum class ZoomLevel( + val level: Int +) { + ZOOM_0(0), + ZOOM_1(1), + ZOOM_2(2), + ZOOM_3(3), + ZOOM_4(4), + ZOOM_5(5), + ; + + val chunkSize: Int = 2f.pow(10 - level).roundToInt() + + init { + require(chunkSize > 0) { "chunkSize must be positive" } + } + + companion object { + val values: List = values().asList() // better performance, KT-48872 + } +} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt index 0843190..010ded3 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt @@ -14,7 +14,7 @@ fun playerUpdatesToWsServer( websocketServer: WebsocketServer, builder: StreamsBuilder = StreamsBuilder(), ) { - builder.stream( + builder.stream( "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}", consumedAs("stream-player-updates-for-ws-server", jsonMapper.serde(), jsonMapper.serde()) ) 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 13e9ae1..30b8839 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 @@ -49,7 +49,7 @@ value class PrototypeName(val name: String) { /** Get the [Colour] of each [MapTilePrototype] */ fun tilePrototypeColourTable(builder: StreamsBuilder): KTable { - return builder.stream( + return builder.stream( "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", consumedAs( "consume-all-prototypes-packets", diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt index 9f6115e..1916e55 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt @@ -108,7 +108,7 @@ fun aggregateWebMapTiles( kxsBinary.serde(), kxsBinary.serde(), ) - ) { pos: TileUpdateRecordKey, px: WebMapTilePixel -> + ) { pos: TileUpdateRecordKey, _: WebMapTilePixel -> val (chunkX, chunkY) = pos.tilePosition.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE) val chunkPos = WebMapTileChunkPosition( WEB_MAP_IMAGE_TILE_SIZE, diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt index 942c517..e5f91c8 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt @@ -14,7 +14,7 @@ import kotlin.math.roundToInt import org.apache.kafka.streams.kstream.KTable -private val COLOUR_TRANSPARENT = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f)) +val COLOUR_TRANSPARENT = RGBColor.fromAwt(Color(0f, 0f, 0f, 0f)) fun saveTileImages( webMapTiles: KTable @@ -54,12 +54,7 @@ private fun saveMapTilesPng( pixels.forEach { (tilePosition, colour) -> - val rgbColour = RGBColor( - colour.red.roundToInt(), - colour.green.roundToInt(), - colour.blue.roundToInt(), - colour.alpha.roundToInt(), - ) + val rgbColour = colour.toRgbColor() val pixelX = abs(abs(tilePosition.x) - abs(chunkOriginX)) val pixelY = abs(abs(tilePosition.y) - abs(chunkOriginY)) diff --git a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt index 1fc21aa..934ebb6 100644 --- a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt +++ b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/map.kt @@ -36,7 +36,7 @@ fun createFactorioMap() = Maps { tms = true // minZoom = -2 // maxZoom = 1 - maxNativeZoom = 1 + maxNativeZoom = 5 minNativeZoom = 1 noWrap = true updateWhenIdle = true diff --git a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt index 2030fe5..ff1d9f1 100644 --- a/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt +++ b/modules/web-map/src/main/kotlin/dev/adamko/kafkatorio/webmap/websockets.kt @@ -46,7 +46,7 @@ class WebsocketService( is ConsoleChatMessage, is EntityData, is MapChunk, - is MapTile, +// is MapTile, is SurfaceData, is MapTiles -> { }