From 560a5d4172a894b7918e3725bdb98c94a9d19551 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Fri, 21 Jan 2022 07:55:36 +0100 Subject: [PATCH] the performance increase wasn't that good actually - so split up the topology into separate files so it's easier to read, and simplifying the png tile process so it joins tiles and prototypes sooner, with less info --- .../events/schema/converters/positions.kt | 32 +- .../processor/KafkatorioTopology.kt | 441 +----------------- .../kafkatorio/processor/WebsocketServer.kt | 4 - .../kafkatorio/processor/eventProcessor.kt | 9 +- .../processor/topology/FactorioGameState.kt | 68 --- .../processor/topology/FactorioServerState.kt | 74 +++ .../processor/topology/factorioMapTiles.kt | 86 ++++ .../topology/factorioServerPackets.kt | 54 +++ .../kafkatorio/processor/topology/model.kt | 10 + .../processor/topology/playerUpdates.kt | 26 ++ .../processor/topology/prototypes.kt | 100 ++++ .../processor/topology/webMapTileAggregate.kt | 125 +++++ .../processor/topology/webMapTiles.kt | 92 ++++ 13 files changed, 603 insertions(+), 518 deletions(-) delete mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioGameState.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioServerState.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt create mode 100644 modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt 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 1cb0ca1..0f80a73 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 @@ -17,20 +17,20 @@ operator fun MapTilePosition.plus(addend: Int) = MapTilePosition(x + addend, y + addend) -fun MapTilePosition.toMapChunkPosition() = +fun MapTilePosition.toMapChunkPosition( chunkSize: Int = MAP_CHUNK_SIZE) = MapChunkPosition( - floor(x.toDouble() / MAP_CHUNK_SIZE.toDouble()).toInt(), - floor(y.toDouble() / MAP_CHUNK_SIZE.toDouble()).toInt(), + floor(x.toDouble() / chunkSize.toDouble()).toInt(), + floor(y.toDouble() / chunkSize.toDouble()).toInt(), ) fun MapEntityPosition.toMapTilePosition() = MapChunkPosition(floor(x).toInt(), floor(y).toInt()) -fun MapEntityPosition.toMapChunkPosition() = +fun MapEntityPosition.toMapChunkPosition( chunkSize: Int = MAP_CHUNK_SIZE) = MapChunkPosition( - floor(x / MAP_CHUNK_SIZE.toDouble()).toInt(), - floor(y / MAP_CHUNK_SIZE.toDouble()).toInt(), + floor(x / chunkSize.toDouble()).toInt(), + floor(y / chunkSize.toDouble()).toInt(), ) @@ -44,15 +44,15 @@ operator fun MapChunkPosition.minus(subtrahend: Int) = MapChunkPosition(x - subtrahend, y - subtrahend) -val MapChunkPosition.leftTopTile: MapTilePosition - get() = MapTilePosition( - x * MAP_CHUNK_SIZE, - y * MAP_CHUNK_SIZE, +fun MapChunkPosition.leftTopTile( chunkSize: Int = MAP_CHUNK_SIZE): MapTilePosition = + MapTilePosition( + x * chunkSize, + y * chunkSize, ) -val MapChunkPosition.rightBottomTile: MapTilePosition - get() = leftTopTile + (MAP_CHUNK_SIZE - 1) - - -val MapChunkPosition.boundingBox: MapBoundingBox - get() = MapBoundingBox(leftTopTile, rightBottomTile) +//val MapChunkPosition.rightBottomTile: MapTilePosition +// get() = leftTopTile + (MAP_CHUNK_SIZE - 1) +// +// +//val MapChunkPosition.boundingBox: MapBoundingBox +// get() = MapBoundingBox(leftTopTile, rightBottomTile) 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 969f136..96cea83 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,64 +1,22 @@ package dev.adamko.kafkatorio.processor -import com.sksamuel.scrimage.ImmutableImage -import com.sksamuel.scrimage.color.RGBColor -import com.sksamuel.scrimage.nio.PngWriter -import dev.adamko.kafkatorio.events.schema.FactorioConfigurationUpdate -import dev.adamko.kafkatorio.events.schema.FactorioEvent -import dev.adamko.kafkatorio.events.schema.FactorioObjectData -import dev.adamko.kafkatorio.events.schema.FactorioPrototypes -import dev.adamko.kafkatorio.events.schema.KafkatorioPacket -import dev.adamko.kafkatorio.events.schema.MAP_CHUNK_SIZE -import dev.adamko.kafkatorio.events.schema.MapChunk -import dev.adamko.kafkatorio.events.schema.MapChunkPosition import dev.adamko.kafkatorio.events.schema.MapTile 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.events.schema.converters.leftTopTile -import dev.adamko.kafkatorio.events.schema.converters.toHexadecimal -import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition -import dev.adamko.kotka.extensions.consumedAs -import dev.adamko.kotka.extensions.groupedAs -import dev.adamko.kotka.extensions.materializedAs -import dev.adamko.kotka.extensions.materializedWith -import dev.adamko.kotka.extensions.producedAs -import dev.adamko.kotka.extensions.streams.aggregate -import dev.adamko.kotka.extensions.streams.filter -import dev.adamko.kotka.extensions.streams.groupBy -import dev.adamko.kotka.extensions.streams.map -import dev.adamko.kotka.extensions.streams.mapValues -import dev.adamko.kotka.extensions.streams.merge -import dev.adamko.kotka.extensions.streams.reduce -import dev.adamko.kotka.extensions.streams.to -import dev.adamko.kotka.extensions.tables.join -import dev.adamko.kotka.extensions.tables.mapValues -import dev.adamko.kotka.kxs.serde -import dev.adamko.kotka.topicdata.TopicRecord -import dev.adamko.kotka.topicdata.flatMapTopicRecords -import java.awt.Color -import java.awt.image.BufferedImage -import java.io.File +import dev.adamko.kafkatorio.processor.topology.PrototypeName +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.playerUpdatesToWsServer +import dev.adamko.kafkatorio.processor.topology.prototypesTable +import dev.adamko.kafkatorio.processor.topology.saveTileImages +import dev.adamko.kafkatorio.processor.topology.splitFactorioServerLog import java.time.Duration -import kotlin.math.abs -import kotlin.math.roundToInt -import kotlinx.serialization.Serializable -import kotlinx.serialization.Transient -import kotlinx.serialization.decodeFromString -import kotlinx.serialization.encodeToString -import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.KafkaStreams -import org.apache.kafka.streams.StoreQueryParameters import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.TopologyDescription -import org.apache.kafka.streams.kstream.Consumed -import org.apache.kafka.streams.kstream.KStream import org.apache.kafka.streams.kstream.KTable -import org.apache.kafka.streams.kstream.Suppressed -import org.apache.kafka.streams.processor.RecordContext -import org.apache.kafka.streams.state.QueryableStoreTypes -import org.apache.kafka.streams.state.ReadOnlyKeyValueStore -import org.apache.kafka.streams.state.ValueAndTimestamp class KafkatorioTopology( @@ -70,21 +28,18 @@ class KafkatorioTopology( const val sourceTopic = "factorio-server-log" } -// private lateinit var tilePrototypesStore: ReadOnlyKeyValueStore> - fun build() { - parseAndSplitServerLog() - sendPlayerUpdatesToWebSocket() -// createMapTileImages() - val tilePrototypesTable: KTable = tilePrototypesTable() + splitFactorioServerLog(builder) + playerUpdatesToWsServer(websocketServer, builder) - val tilesTable: KTable = buildTilesTable() + val allMapTilesTable: KTable = allMapTilesTable(builder) + val tilePrototypesTable: KTable = prototypesTable(builder) - val tilesGroupedByChunk: KTable = - buildChunkedTilesTable(tilesTable) + val webMapTiles: KTable = + aggregateWebMapTiles(allMapTilesTable, tilePrototypesTable) - createChunkTileImage(tilePrototypesTable, tilesGroupedByChunk) + saveTileImages(webMapTiles) val topology = builder.build() @@ -99,370 +54,6 @@ class KafkatorioTopology( streams.start() -// tilePrototypesStore = streams.store( -// StoreQueryParameters.fromNameAndType( -// tilePrototypesTable.queryableStoreName(), -// QueryableStoreTypes.timestampedKeyValueStore() -// ) -// ) - } - - private fun parseAndSplitServerLog() { - - builder.stream( - sourceTopic, - consumedAs( - "read-raw-packets-from-server", - Serdes.String(), - Serdes.String(), - ) - ) - .mapValues("decode-packets") { _: String, value: String -> -// println("Mapping $readOnlyKey:$value") - jsonMapper.decodeFromString(value) - } - .to( - producedAs( - "split-server-log", - Serdes.String(), - KafkatorioPacketSerde - ) - ) { _: String, value: KafkatorioPacket, _: RecordContext -> -// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}") - when (value) { - is FactorioEvent -> - "kafkatorio.${value.packetType.name}.${value.data.objectName.name}" - is FactorioConfigurationUpdate -> - "kafkatorio.${value.packetType.name}.FactorioConfigurationUpdate" - is FactorioPrototypes -> - "kafkatorio.${value.packetType.name}.all" - } - } - - } - - private fun sendPlayerUpdatesToWebSocket() { - builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}", - Consumed.with(Serdes.String(), KafkatorioPacketSerde) - ) - .foreach { _, value -> -// println("sending ${value.packetType} packet to websocket") - websocketServer.sendMessage(jsonMapper.encodeToString(value)) - } - } - - @Serializable - data class SurfaceIndex( - val surfaceIndex: Int - ) { - override fun toString() = "$surfaceIndex" - } - - @Serializable - data class SurfacePrototypesData( - val surfaceIndex: SurfaceIndex, - val mapTilePrototypes: Set, - ) - - private fun tilePrototypesTable(): KTable { - -// val tilePrototypesStreamName = "kafkatorio.stream.prototypes-tiles" -// val tilePrototypesStoreName = "kafkatorio.store.prototypes-tiles" - - return builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", - consumedAs( - "consume-all-prototypes-packets", - Serdes.String(), - jsonMapper.serde() - ) - ) - .map("flatten-map-tile-prototypes") { _, prototypes: FactorioPrototypes -> - val tilePrototypes = prototypes.prototypes - .filterIsInstance() - .toSet() - val key = SurfaceIndex(1) // TODO make surfaceIndex come from Factorio, not hard coded - val data = SurfacePrototypesData(key, tilePrototypes) - key to data - } - .peek { _, value -> - println("tile prototypes: ${value.mapTilePrototypes.joinToString { it.name }}") - } - .groupByKey( - groupedAs( - "grouping-tile-prototypes-by-surface-index", - jsonMapper.serde(), - jsonMapper.serde() - ) - ) - .reduce( - "group-surface-prototypes", - materializedWith( -// "kafkatorio.surface-prototypes", - jsonMapper.serde(), - jsonMapper.serde() - ) - ) { value1: SurfacePrototypesData, value2: SurfacePrototypesData -> - value1.copy(mapTilePrototypes = value1.mapTilePrototypes + value2.mapTilePrototypes) - } -// .toTable( -// materializedAs( -// "kafkatorio.prototypes.map-tiles", -// jsonMapper.serde(), -// jsonMapper.serde() -// ) -// ) -// .to( -// tilePrototypesStreamName, -// ) - -// return builder.globalTable( -// tilePrototypesStreamName, -// consumedAs( -// "send-tile-protos-to-gkt-store", -// jsonMapper.serde(), -// jsonMapper.serde() -// ), -// materializedAs( -// storeName = tilePrototypesStoreName, -// keySerde = jsonMapper.serde(), -// valueSerde = jsonMapper.serde(), -// ) -// ) - } - - @Serializable - data class TileUpdateRecordKey( - val surfaceIndex: SurfaceIndex, - val tilePosition: MapTilePosition, - ) - - @Serializable - data class TileUpdateRecord( - val surfaceIndex: SurfaceIndex, - val tilePosition: MapTilePosition, - val tile: MapTile, - ) : TopicRecord { - @Transient - override val topicKey: TileUpdateRecordKey = TileUpdateRecordKey(surfaceIndex, tilePosition) - } - - private fun buildTilesTable(): KTable { - - val luaTilesUpdatesStream: KStream = builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", - consumedAs("consume-map-tiles-packets", Serdes.String(), jsonMapper.serde()) - ) - .filter("events.filter.map-tiles") { _: String, event: FactorioEvent -> event.data is MapTiles } - .mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! } - - val chunkTilesUpdateStream: KStream = builder.stream( - "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", - consumedAs("consume-map-chunk-packets", Serdes.String(), jsonMapper.serde()) - ) - .filter("events.filter.map-chunks") { _: String, event: FactorioEvent -> event.data is MapChunk } - .mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles } - - return luaTilesUpdatesStream - .merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream) - .flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles -> -// println("all map tiles update ${mapTiles.tiles.size}") - mapTiles.tiles.map { tile -> - TileUpdateRecord( - SurfaceIndex(mapTiles.surfaceIndex), - tile.position, - tile, - ) - } - } - .toTable(materializedWith(jsonMapper.serde(), jsonMapper.serde())) -// .suppress( -// Suppressed.untilTimeLimit( -// Duration.ofSeconds(30), -// Suppressed.BufferConfig -// .maxRecords(100) -// .emitEarlyWhenFull() -// ) -// ) - } - - @Serializable - data class MapChunkDataPosition( - val position: MapChunkPosition, - val surfaceIndex: SurfaceIndex, - ) - - @Serializable - data class MapChunkData( - val chunkPosition: MapChunkDataPosition, - val tiles: Set, - ) - - private fun buildChunkedTilesTable( - tilesTable: KTable - ): KTable { - - @Serializable - data class MapChunkDataAggregator( - val tiles: MutableSet = mutableSetOf(), - ) - - return tilesTable - .toStream() -// .repartition( -// repartitionedAs( -// "repartition-all-tiles-before-chunk-grouping", -// jsonMapper.serde(), -// jsonMapper.serde() -// ) -// ) - .groupBy( - groupedAs("grouping-tiles-by-chunk-pos", jsonMapper.serde(), jsonMapper.serde()), - ) { key: TileUpdateRecordKey, value: TileUpdateRecord -> - MapChunkDataPosition( - value.tilePosition.toMapChunkPosition(), - key.surfaceIndex - ) - } - .aggregate( - "reducing-tiles-grouped-by-chunk-position", - materializedWith(jsonMapper.serde(), jsonMapper.serde()), - { MapChunkDataAggregator() }, - ) { _: MapChunkDataPosition, value: TileUpdateRecord, aggregate: MapChunkDataAggregator -> - aggregate.tiles.add(value.tile) - aggregate - } - // TODO maybe set serdes here? - .mapValues( - "tiles-grouped-by-chunk-position-finalise-aggregation", - materializedWith(jsonMapper.serde(), jsonMapper.serde()), - ) { k, v -> - MapChunkData(k, v.tiles) - } - - } - - private fun createChunkTileImage( - tilePrototypesTable: KTable, - tilesGroupedByChunk: KTable, - ) { - - @Serializable - data class ChunkTilesAndProtos( - val chunkTiles: MapChunkData, - val protos: SurfacePrototypesData - ) - -// tilesGroupedByChunk -// .join( -// tilePrototypesTable, -// { cd: MapChunkData -> cd.chunkPosition.surfaceIndex }, -// { chunkTiles: MapChunkData, protos: SurfacePrototypesData -> -// ChunkTilesAndProtos(chunkTiles, protos) -// }, -// namedAs("joining-chunks-tiles-prototypes"), -// materializedAs( -// "joined-chunked-tiles-with-prototypes", -// jsonMapper.serde(), -// jsonMapper.serde() -// ), -// ) - tilesGroupedByChunk - .join( - tilePrototypesTable, - "enrich-tiles-with-prototype-colour", - materializedAs( - "joined-chunked-tiles-with-prototypes", - jsonMapper.serde(), - jsonMapper.serde() - ), - { it.chunkPosition.surfaceIndex }, - ) { chunkTiles: MapChunkData, protos: SurfacePrototypesData -> - ChunkTilesAndProtos(chunkTiles, protos) - } - .suppress( - Suppressed.untilTimeLimit( - Duration.ofSeconds(30), - Suppressed.BufferConfig.maxRecords(30) - ) - ) - .toStream() - .foreach { _, (chunkTiles: MapChunkData, protos: SurfacePrototypesData) -> - runCatching { - saveMapTilesPng(chunkTiles, protos) - }.onFailure { e -> - println("error saving map tile png chunk:${chunkTiles.chunkPosition}") - e.printStackTrace() - throw e - } - } - } - - - private fun saveMapTilesPng(chunkTiles: MapChunkData, protos: SurfacePrototypesData) { - - val chunkPos = chunkTiles.chunkPosition - - val tilePrototypes: Map = - protos.mapTilePrototypes.associateBy { it.name }.withDefault { null } - - val chunkOriginX = chunkTiles.chunkPosition.position.leftTopTile.x -// val chunkOriginY = chunk.chunkPosition.position.rightBottomTile.y - val chunkOriginY = chunkTiles.chunkPosition.position.leftTopTile.y - - val chunkImage = - ImmutableImage.filled( - MAP_CHUNK_SIZE, - MAP_CHUNK_SIZE, - Color.BLACK, - BufferedImage.TYPE_INT_ARGB - ) - - chunkTiles.tiles.forEach { tile -> - - val prototypeColour = tilePrototypes - .getValue(tile.prototypeName) - ?.mapColour - ?.toHexadecimal() - - val rgbColour = when (prototypeColour) { - null -> { - println("missing prototype: ${tile.prototypeName}") - RGBColor(11, 11, 11, 0) - } - else -> RGBColor( - prototypeColour.red.roundToInt(), - prototypeColour.green.roundToInt(), - prototypeColour.blue.roundToInt(), - prototypeColour.alpha.roundToInt(), - ) - } - - val pixelX = abs(abs(tile.position.x) - abs(chunkOriginX)) - val pixelY = abs(abs(tile.position.y) - abs(chunkOriginY)) - - chunkImage.setColor( - pixelX, - pixelY, - rgbColour - ) - - } - - val zoom = 1u - val file = - File( - "src/main/resources/kafkatorio-web-map/s${chunkPos.surfaceIndex}/z$zoom/x${chunkOriginX}/y${chunkOriginY}.png" - ) - - if (file.parentFile.mkdirs()) { - println("created new map tile parentFile directory ${file.absolutePath}") - } - - println("saving map tile $file") - - chunkImage.output(PngWriter.MaxCompression, file) } } diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt index d573ded..99721bb 100644 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/WebsocketServer.kt @@ -10,12 +10,8 @@ import kotlinx.coroutines.flow.withIndex import kotlinx.coroutines.job import kotlinx.coroutines.runBlocking import kotlinx.coroutines.supervisorScope -import org.http4k.core.Request -import org.http4k.core.Response -import org.http4k.core.Status.Companion.OK import org.http4k.routing.bind import org.http4k.routing.websockets -import org.http4k.server.PolyHandler import org.http4k.websocket.Websocket import org.http4k.websocket.WsHandler import org.http4k.websocket.WsMessage 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 faaf92b..634fb6b 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 @@ -3,7 +3,6 @@ package dev.adamko.kafkatorio.processor import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer -import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import org.http4k.server.PolyHandler import org.http4k.server.Undertow @@ -24,11 +23,11 @@ fun main() = runBlocking { wsServer.build(), ).asServer(Undertow(9073)) -// Runtime.getRuntime().addShutdownHook(Thread { -// webServer.stop() -// }) + Runtime.getRuntime().addShutdownHook(Thread { + webServer.stop() + }) // launch { - webServer.start().block() + webServer.start().block() // } } 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 deleted file mode 100644 index d66e7d1..0000000 --- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioGameState.kt +++ /dev/null @@ -1,68 +0,0 @@ -package dev.adamko.kafkatorio.processor.topology - -import dev.adamko.kafkatorio.events.schema.KafkatorioPacket -import dev.adamko.kafkatorio.events.schema.MapChunkPosition -import dev.adamko.kafkatorio.events.schema.MapTilePosition -import dev.adamko.kafkatorio.events.schema.MapTilePrototype -import dev.adamko.kafkatorio.processor.jsonMapper -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.Consumed -import org.apache.kafka.streams.kstream.KGroupedStream - -class FactorioGameState ( -private val builder: StreamsBuilder = StreamsBuilder(), -) { - private val sourceTopic = "factorio-server-log" - - - private fun gatherGameState() { - - val serverPackets: KGroupedStream = - builder.stream( - sourceTopic, - Consumed.with(Serdes.String(), Serdes.String()) - ) - .mapValues { readOnlyKey, value -> - println("Mapping $readOnlyKey:$value") - jsonMapper.decodeFromString(value) - } - .groupByKey() - - @Serializable - data class TileSurfaceData( - val surfaceIndex: Int, - val position: MapTilePosition, - val prototype: MapTilePrototype, - ) - - @Serializable - data class MapChunkData( - val position: MapChunkPosition, - val tiles: List, - ) - - @Serializable - data class FactorioSurfaceData( - val surfaceIndex: Int, - val chunks: List, - ) - - @Serializable - data class FactorioServerState( - val surfaces: Map, - val tilePrototypes: List, - ) - - - -// serverPackets -// .aggregate( -// Initializer { FactorioServerState("", emptyMap(), emptyList()) } -// ) - } - - -} diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioServerState.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioServerState.kt new file mode 100644 index 0000000..48a80b0 --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/FactorioServerState.kt @@ -0,0 +1,74 @@ +package dev.adamko.kafkatorio.processor.topology + +import dev.adamko.kafkatorio.events.schema.KafkatorioPacket +import dev.adamko.kafkatorio.events.schema.MapChunkPosition +import dev.adamko.kafkatorio.events.schema.MapTilePosition +import dev.adamko.kafkatorio.events.schema.MapTilePrototype +import dev.adamko.kafkatorio.processor.jsonMapper +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.Consumed +import org.apache.kafka.streams.kstream.KGroupedStream + +private const val sourceTopic = "factorio-server-log" + +@JvmInline +@Serializable +value class FactorioServerId(val id: String) { + override fun toString() = id +} + +@Serializable +data class FactorioServerState( + val serverId: FactorioServerId, + val mapTilePrototypes: Set, + val surfaceIndexes: Set, +) + + +private fun gatherGameState(builder: StreamsBuilder = StreamsBuilder()) { + + val serverPackets: KGroupedStream = + builder.stream( + sourceTopic, + Consumed.with(Serdes.String(), Serdes.String()) + ) + .mapValues { readOnlyKey, value -> + println("Mapping $readOnlyKey:$value") + jsonMapper.decodeFromString(value) + } + .groupByKey() + + @Serializable + data class TileSurfaceData( + val surfaceIndex: Int, + val position: MapTilePosition, + val prototype: MapTilePrototype, + ) + + @Serializable + data class MapChunkData( + val position: MapChunkPosition, + val tiles: List, + ) + + @Serializable + data class FactorioSurfaceData( + val surfaceIndex: Int, + val chunks: List, + ) + + @Serializable + data class FactorioServerState( + val surfaces: Map, + val tilePrototypes: List, + ) + + +// serverPackets +// .aggregate( +// Initializer { FactorioServerState("", emptyMap(), emptyList()) } +// ) +} 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 new file mode 100644 index 0000000..de01bd0 --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioMapTiles.kt @@ -0,0 +1,86 @@ +package dev.adamko.kafkatorio.processor.topology + +import dev.adamko.kafkatorio.events.schema.FactorioEvent +import dev.adamko.kafkatorio.events.schema.FactorioObjectData +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.MapTilePosition +import dev.adamko.kafkatorio.events.schema.MapTiles +import dev.adamko.kafkatorio.processor.jsonMapper +import dev.adamko.kotka.extensions.consumedAs +import dev.adamko.kotka.extensions.materializedWith +import dev.adamko.kotka.extensions.streams.filter +import dev.adamko.kotka.extensions.streams.flatMap +import dev.adamko.kotka.extensions.streams.mapValues +import dev.adamko.kotka.extensions.streams.toTable +import dev.adamko.kotka.kxs.serde +import java.time.Duration +import kotlinx.serialization.Serializable +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.kstream.KTable +import org.apache.kafka.streams.kstream.Suppressed + +@Serializable +data class TileUpdateRecordKey( + val surfaceIndex: SurfaceIndex, + val tilePosition: MapTilePosition, +) + + +fun allMapTilesTable( + builder: StreamsBuilder +): KTable { + + val luaTilesUpdatesStream: KStream = builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaTiles}", + consumedAs("consume-map-tiles-packets", Serdes.String(), jsonMapper.serde()) + ) + .filter("events.filter.map-tiles") { _: String, event: FactorioEvent -> event.data is MapTiles } + .mapValues("events.extract-map-tiles") { _, event: FactorioEvent -> (event.data as? MapTiles)!! } + + val chunkTilesUpdateStream: KStream = builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.MapChunk}", + consumedAs("consume-map-chunk-packets", Serdes.String(), jsonMapper.serde()) + ) + .filter("events.filter.map-chunks") { _: String, event: FactorioEvent -> event.data is MapChunk } + .mapValues("events.extract-map-chunks") { _, event: FactorioEvent -> (event.data as? MapChunk)!!.tiles } + + return luaTilesUpdatesStream + .merge(chunkTilesUpdateStream) + .flatMap("flat-map-all-tiles") { _, value -> + value.tiles.map { it.position to it } + } + .toTable( + "all-tiles-keyed-on-position", + materializedWith(jsonMapper.serde(), jsonMapper.serde()) + ) + .suppress( + Suppressed.untilTimeLimit( + Duration.ofSeconds(30), + Suppressed.BufferConfig + .maxRecords(30) + .emitEarlyWhenFull() + ) + ) +} + +//fun buildTilesTable( +// builder: StreamsBuilder +//): KTable { +// +// return luaTilesUpdatesStream +// .merge("luaTilesUpdatesStream-and-chunkTilesUpdateStream", chunkTilesUpdateStream) +// .flatMapTopicRecords("all-tiles-convert-to-TileUpdateRecord") { _, mapTiles: MapTiles -> +// mapTiles.tiles.map { tile -> +// TileUpdateRecord( +// SurfaceIndex(mapTiles.surfaceIndex), +// tile.position, +// tile, +// ) +// } +// } +// .toTable(materializedWith(jsonMapper.serde(), jsonMapper.serde())) +//} 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 new file mode 100644 index 0000000..d071ad2 --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt @@ -0,0 +1,54 @@ +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.processor.KafkatorioPacketSerde +import dev.adamko.kafkatorio.processor.KafkatorioTopology +import dev.adamko.kafkatorio.processor.jsonMapper +import dev.adamko.kotka.extensions.consumedAs +import dev.adamko.kotka.extensions.producedAs +import dev.adamko.kotka.extensions.streams.mapValues +import dev.adamko.kotka.extensions.streams.to +import kotlinx.serialization.decodeFromString +import org.apache.kafka.common.serialization.Serdes +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.processor.RecordContext + + +fun splitFactorioServerLog( + builder: StreamsBuilder = StreamsBuilder(), +) { + + builder.stream( + KafkatorioTopology.sourceTopic, + consumedAs( + "read-raw-packets-from-server", + Serdes.String(), + Serdes.String(), + ) + ) + .mapValues("decode-packets") { _: String, value: String -> +// println("Mapping $readOnlyKey:$value") + jsonMapper.decodeFromString(value) + } + .to( + producedAs( + "split-server-log", + Serdes.String(), + KafkatorioPacketSerde + ) + ) { _: String, value: KafkatorioPacket, _: RecordContext -> +// println("[$key] sending event:${value.eventType} to topic:${value.data.objectName()}") + when (value) { + is FactorioEvent -> + "kafkatorio.${value.packetType.name}.${value.data.objectName.name}" + is FactorioConfigurationUpdate -> + "kafkatorio.${value.packetType.name}.FactorioConfigurationUpdate" + is FactorioPrototypes -> + "kafkatorio.${value.packetType.name}.all" + } + } + +} 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 new file mode 100644 index 0000000..1ba0d5c --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/model.kt @@ -0,0 +1,10 @@ +package dev.adamko.kafkatorio.processor.topology + +import kotlinx.serialization.Serializable + +@Serializable +data class SurfaceIndex( + val surfaceIndex: Int +) { + override fun toString() = "$surfaceIndex" +} 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 new file mode 100644 index 0000000..da1838c --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/playerUpdates.kt @@ -0,0 +1,26 @@ +package dev.adamko.kafkatorio.processor.topology + +import dev.adamko.kafkatorio.events.schema.FactorioObjectData +import dev.adamko.kafkatorio.events.schema.KafkatorioPacket +import dev.adamko.kafkatorio.processor.KafkatorioPacketSerde +import dev.adamko.kafkatorio.processor.WebsocketServer +import dev.adamko.kafkatorio.processor.jsonMapper +import kotlinx.serialization.encodeToString +import org.apache.kafka.common.serialization.Serdes +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.kstream.Consumed + + +fun playerUpdatesToWsServer( + websocketServer: WebsocketServer, + builder: StreamsBuilder = StreamsBuilder(), +) { + builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.EVENT}.${FactorioObjectData.ObjectName.LuaPlayer}", + Consumed.with(Serdes.String(), KafkatorioPacketSerde) + ) + .foreach { _, value -> +// println("sending ${value.packetType} packet to websocket") + websocketServer.sendMessage(jsonMapper.encodeToString(value)) + } +} 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 new file mode 100644 index 0000000..a09c5b7 --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/prototypes.kt @@ -0,0 +1,100 @@ +package dev.adamko.kafkatorio.processor.topology + +import dev.adamko.kafkatorio.events.schema.FactorioPrototypes +import dev.adamko.kafkatorio.events.schema.KafkatorioPacket +import dev.adamko.kafkatorio.events.schema.MapTilePrototype +import dev.adamko.kafkatorio.processor.jsonMapper +import dev.adamko.kotka.extensions.consumedAs +//import dev.adamko.kotka.extensions.groupedAs +import dev.adamko.kotka.extensions.materializedWith +import dev.adamko.kotka.extensions.streams.flatMap +//import dev.adamko.kotka.extensions.streams.* +import dev.adamko.kotka.kxs.serde +import kotlinx.serialization.Serializable +import org.apache.kafka.common.serialization.Serdes +import org.apache.kafka.streams.StreamsBuilder +import org.apache.kafka.streams.kstream.KTable + + +@Serializable +@JvmInline +value class PrototypeName(val name: String) { + override fun toString() = name +} + + +fun prototypesTable(builder: StreamsBuilder): KTable { + return builder.stream( + "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", + consumedAs( + "consume-all-prototypes-packets", + Serdes.String(), + jsonMapper.serde() + ) + ) + .flatMap("map-MapTilePrototype") { _, prototypes: FactorioPrototypes -> + prototypes.prototypes + .filterIsInstance() + .map { PrototypeName(it.name) to it } + } + .toTable( + materializedWith( + jsonMapper.serde(), + jsonMapper.serde() + ) + ) +} + + +//@Serializable +//data class FactorioServerPrototypesRecord( +// val serverId: FactorioServerId, +// val mapTilePrototypes: Map, +//) +// +//fun serverPrototypesTable(builder: StreamsBuilder): KTable { +// +// return builder.stream( +// "kafkatorio.${KafkatorioPacket.PacketType.PROTOTYPES}.all", +// consumedAs( +// "consume-all-prototypes-packets", +// Serdes.String(), +// jsonMapper.serde() +// ) +// ) +// .map("build-FactorioServerPrototypesRecord", ::buildPrototypesRecord) +// .peek { _, value -> +// println("tile prototypes: ${value.mapTilePrototypes.keys.joinToString()}") +// } +// .groupByKey( +// groupedAs( +// "grouping-tile-prototypes-by-surface-index", +// jsonMapper.serde(), +// jsonMapper.serde() +// ) +// ) +// .reduce( +// "group-surface-prototypes", +// materializedWith( +//// "kafkatorio.surface-prototypes", +// jsonMapper.serde(), +// jsonMapper.serde() +// ) +// ) { value1: FactorioServerPrototypesRecord, value2: FactorioServerPrototypesRecord -> +// val allTiles = value1.mapTilePrototypes + value2.mapTilePrototypes +// value1.copy(mapTilePrototypes = allTiles) +// } +//} +// +//private fun buildPrototypesRecord( +// serverId: String, +// prototypes: FactorioPrototypes +//): Pair { +// val tilePrototypes = prototypes.prototypes +// .filterIsInstance() +// .associateBy { PrototypeName(it.name) } +// +// val key = FactorioServerId(serverId) +// val data = FactorioServerPrototypesRecord(key, tilePrototypes) +// return key to data +//} 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 new file mode 100644 index 0000000..954a46c --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTileAggregate.kt @@ -0,0 +1,125 @@ +package dev.adamko.kafkatorio.processor.topology + +import dev.adamko.kafkatorio.events.schema.Colour +import dev.adamko.kafkatorio.events.schema.MapTile +import dev.adamko.kafkatorio.events.schema.MapTilePosition +import dev.adamko.kafkatorio.events.schema.MapTilePrototype +import dev.adamko.kafkatorio.events.schema.converters.toMapChunkPosition +import dev.adamko.kafkatorio.processor.jsonMapper +import dev.adamko.kotka.extensions.groupedAs +import dev.adamko.kotka.extensions.materializedAs +import dev.adamko.kotka.extensions.namedAs +import dev.adamko.kotka.extensions.tables.groupBy +import dev.adamko.kotka.extensions.tables.join +import dev.adamko.kotka.extensions.tables.mapValues +import dev.adamko.kotka.extensions.toKeyValue +import dev.adamko.kotka.kxs.serde +import java.time.Duration +import kotlinx.serialization.Serializable +import org.apache.kafka.streams.kstream.KTable +import org.apache.kafka.streams.kstream.Suppressed + +/** The tiles of the webmap will be this many pixels high/wide */ +const val WEB_MAP_IMAGE_TILE_SIZE = 32 + +//@Serializable +//data class MapChunkDataPosition( +// val position: MapChunkPosition, +// val surfaceIndex: SurfaceIndex, +//) +// +//@Serializable +//data class MapChunkData( +// val chunkPosition: MapChunkDataPosition, +// val tiles: Set, +//) + +@Serializable +data class WebMapTileChunkPosition( + val chunkSize: Int, + val x: Int, + val y: Int, +) + +@Serializable +data class WebMapTilePixel( + val mapColour: Colour, + val tilePosition: MapTilePosition, +) + +@Serializable +data class WebMapTileChunkPixels( + val pixels: Set = mutableSetOf(), +) + +/** Temporary mutable aggregator */ +@Serializable +private data class WebMapTileChunkPixelsAcc( + val pixels: MutableSet = mutableSetOf(), +) + +fun aggregateWebMapTiles( + allMapTilesTable: KTable, + tilePrototypesTable: KTable, +): KTable { + + val webmapTileColours: KTable = + allMapTilesTable + .join( + tilePrototypesTable, + "convert-tiles-to-colours", + materializedAs( + "store-web-map-tile-pixels", + jsonMapper.serde(), + jsonMapper.serde(), + ), + { PrototypeName(it.prototypeName) }, + ) { tile: MapTile, proto: MapTilePrototype -> + WebMapTilePixel(proto.mapColour, tile.position) + } + + + val webMapTileChunkAggregate: KTable = + webmapTileColours + .groupBy( + groupedAs( + "group-webmap-tile-colours-into-chunks", + jsonMapper.serde(), + jsonMapper.serde(), + ) + ) { pos: MapTilePosition, px: WebMapTilePixel -> + val (chunkX, chunkY) = pos.toMapChunkPosition(WEB_MAP_IMAGE_TILE_SIZE) + val chunkPos = WebMapTileChunkPosition(WEB_MAP_IMAGE_TILE_SIZE, chunkX, chunkY) + (chunkPos to px).toKeyValue() + } + .aggregate( + { WebMapTileChunkPixelsAcc() }, + /* adder */ + { _: WebMapTileChunkPosition, newValue: WebMapTilePixel, acc: WebMapTileChunkPixelsAcc -> + acc.apply { pixels.add(newValue) } + }, + /* subtractor */ + { _: WebMapTileChunkPosition, oldValue: WebMapTilePixel, acc: WebMapTileChunkPixelsAcc -> + acc.apply { pixels.minus(oldValue) } + }, + namedAs("web-map-tile-colours-into-$WEB_MAP_IMAGE_TILE_SIZE-chunks"), + materializedAs("web-map-tile-colour-chunks-aggregate", jsonMapper.serde(), jsonMapper.serde()) + ) + .suppress( + Suppressed.untilTimeLimit( + Duration.ofSeconds(30), + Suppressed.BufferConfig + .maxRecords(2) +// .withMaxBytes(31457280 ) // 30MB + .emitEarlyWhenFull() + ).withName("web-map-tile-aggregate-debounce") + ) + + return webMapTileChunkAggregate + .mapValues( + "finalise-web-map-tile-colour-chunk-aggregation", + materializedAs("web-map-tile-colour-chunks", jsonMapper.serde(), jsonMapper.serde()) + ) { _, v -> + WebMapTileChunkPixels(v.pixels) + } +} 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 new file mode 100644 index 0000000..5d8b231 --- /dev/null +++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/webMapTiles.kt @@ -0,0 +1,92 @@ +package dev.adamko.kafkatorio.processor.topology + +import com.sksamuel.scrimage.ImmutableImage +import com.sksamuel.scrimage.color.RGBColor +import com.sksamuel.scrimage.nio.PngWriter +import dev.adamko.kafkatorio.events.schema.MAP_CHUNK_SIZE +import dev.adamko.kotka.extensions.tables.toStream +import java.awt.Color +import java.awt.image.BufferedImage +import java.io.File +import java.time.Duration +import kotlin.math.abs +import kotlin.math.roundToInt +import org.apache.kafka.streams.kstream.KTable +import org.apache.kafka.streams.kstream.Suppressed + + +fun saveTileImages( + webMapTiles: KTable +) { + + webMapTiles + .suppress( + Suppressed.untilTimeLimit( + Duration.ofSeconds(30), + Suppressed.BufferConfig.maxRecords(30) + ) + ) + .toStream("get-webmap-tile-updates") + .foreach { position: WebMapTileChunkPosition, pixels: WebMapTileChunkPixels -> + runCatching { + saveMapTilesPng(position, pixels.pixels) + }.onFailure { e -> + println("error saving map tile png chunk: $position") + e.printStackTrace() + throw e + } + } +} + + +private fun saveMapTilesPng( + chunkPosition: WebMapTileChunkPosition, + pixels: Set +) { + val surfaceIndex = 1 // TODO get surface index + + val chunkOriginX = chunkPosition.x * chunkPosition.chunkSize + val chunkOriginY = chunkPosition.y * chunkPosition.chunkSize + + val chunkImage = + ImmutableImage.filled( + MAP_CHUNK_SIZE, + MAP_CHUNK_SIZE, + Color.BLACK, + BufferedImage.TYPE_INT_ARGB + ) + + pixels.forEach { pixel -> + + val rgbColour = RGBColor( + pixel.mapColour.red.roundToInt(), + pixel.mapColour.green.roundToInt(), + pixel.mapColour.blue.roundToInt(), + pixel.mapColour.alpha.roundToInt(), + ) + + val pixelX = abs(abs(pixel.tilePosition.x) - abs(chunkOriginX)) + val pixelY = abs(abs(pixel.tilePosition.y) - abs(chunkOriginY)) + + chunkImage.setColor( + pixelX, + pixelY, + rgbColour + ) + + } + + val zoom = 1u + val file = + File( + "src/main/resources/kafkatorio-web-map/s${surfaceIndex}/z$zoom/x${chunkOriginX}/y${chunkOriginY}.png" + ) + + if (file.parentFile.mkdirs()) { + println("created new map tile parentFile directory ${file.absolutePath}") + } + + println("saving map tile $file") + + chunkImage.output(PngWriter.MaxCompression, file) +}