SPATEM receive, RSU CAM decode, and two ASN.1 encoding fixes
SPATEM over the air - gn_unwrap.c accepts BTP-B port 2004 alongside 2001/2002. The serial protocol already carries the port in its V2X_RX prefix, so nothing else changed there. Note the crossover that makes this easy to get wrong: SPATEM is port 2004 but messageID 4, while MAPEM is port 2003 and messageID 5. - SpatemUperCodec decodes SPAT down to per-signal-group phase and timing. The bit layout was validated by replaying 79,042 real SPATEMs - the whole 2026-03-18 drive across 7+ RSUs plus the bench trigger - against asn1tools using the ETSI modules. All 79,042 matched on every field, none hit an unsupported branch. Two traps are pinned by tests: TimeChangeDetails is the one SEQUENCE here that is NOT extensible (5 optional bits, no extension bit), and maneuverAssistList cannot be skipped when present - it is variable-length, so it has to be walked to find where the next movement starts. - The V2X list shows one row per intersection with each signal group coloured by phase and a countdown where the RSU supplies timing. TimeMark wraps hourly, so the countdown corrects for it; without that it reads hugely negative once an hour, precisely when someone is watching it. - Entries expire after 15 s, much shorter than DENM's window: a traffic light that stopped updating is not "still green". Size caveat, deliberately deferred: SERIAL_LINK_MAX_PAYLOAD is still 512, so a SPATEM over ~498 bytes is counted as an oversize drop. The bench RSU sends 58 bytes and is unaffected, but real road RSUs measured 555 median / 1243 max, so roughly 70% would not arrive. Raising the cap also requires enlarging RX_FRAME_MAX_LEN and moving rx_item_t off the WiFi driver's callback stack, where it would otherwise overflow. RSU CAM decode - HighFrequencyContainer is a CHOICE, and a roadside unit picks rsuContainerHighFrequency, which carries no kinematics at all. The decoder bailed on that branch, so every RSU CAM was dropped - including the bench RSU, which sends CAM and SPATEM from the same station id. It now decodes for position and stationType. - RSU CAMs are kept out of UseCaseDetectionEngine. They arrive as a permanently stationary station at a fixed point, which is exactly the shape the stopped-vehicle and intersection-movement use cases match, and would raise a standing false alert for as long as the RSU was in range. CAM transmit: yawRateConfidence - YawRateConfidence has nine enumerands (0..8), so UPER needs 4 bits and "unavailable" is 8. The encoder wrote 3 bits with value 7 - one bit short and the wrong symbol - shifting every field after yawRate for any standards-strict receiver. The decoder read 3 bits too, so phone and ESP32 agreed with each other and with nothing else. - This is the third instance of that exact failure mode in this project, after CurvatureCalculationMode and the GeoNetworking reserved bytes. A round-trip test through our own decoder structurally cannot catch it, so CamEncodeGolden Test asserts the bytes asn1tools produces instead: it decoded this encoder's output and re-encoded it byte-identically. Confirmed on air afterwards - 26 of our own CAMs captured back off the OBU's receiver, all 26 accepted, where the same decoder rejected them before. DENM - Hazards now expire 60 s after their last repetition. This needs a clock, not just a filter: both source flows only emit when a DENM arrives, so a sender that drives away or loses power would never trigger a recompute and its hazard would stay on screen indefinitely. - The MQTT path was dropping every DENM for two independent reasons, both found by checking the payload against CI-CiT-MQTT_API_Documentation-v6 listing 2.6 rather than guessing: the station id key is originatingStationId, and eventPosition IS a GeoJSON Point rather than an object containing one. Also parses termination (presence is the signal), sequenceNumber, stationType and the RFC3339 detectionTime. Note roadSideUnit is 15, not 12 - the enumeration has a gap after tram(11). V2X screen - The decoded CAM/DENM list now renders on the CiT One path too; it was gated to the ESP32-C5 path and CiT One fell through to the raw MQTT topic list. Those topics move to their own tab, hidden on the ESP32-C5 path where there is no broker. Testing - Adds org.json as a test-only dependency: the android.jar stub throws "not mocked" on every JSONObject call, which made the MQTT payload parsers untestable off-device. - 23 V2X tests pass. EventDetectorTest's 4 failures are pre-existing and untouched by this change.
This commit is contained in:
Ashin Walpola
@@ -17,13 +17,17 @@ import com.hawhamburg.micr0bu.data.transport.UsbSerialState
|
||||
import com.hawhamburg.micr0bu.data.transport.UsbSerialTransport
|
||||
import com.hawhamburg.micr0bu.domain.denm.DenmEvent
|
||||
import com.hawhamburg.micr0bu.domain.denm.DenmParser
|
||||
import com.hawhamburg.micr0bu.domain.spat.SpatIntersection
|
||||
import com.hawhamburg.micr0bu.domain.denm.DenmUseCase
|
||||
import com.hawhamburg.micr0bu.domain.usecase.UseCaseAlert
|
||||
import com.hawhamburg.micr0bu.domain.usecase.UseCaseType
|
||||
import com.hawhamburg.micr0bu.service.CamPinger
|
||||
import dagger.hilt.android.lifecycle.HiltViewModel
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.combine
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.flow
|
||||
import kotlinx.coroutines.flow.runningFold
|
||||
import kotlinx.coroutines.flow.SharingStarted
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
@@ -155,17 +159,77 @@ class MqttViewModel @Inject constructor(
|
||||
camUseCaseRepository.airDenm
|
||||
.runningFold(emptyMap<String, DenmEvent>()) { acc, denm -> acc + (denm.dedupKey to denm) }
|
||||
.map { it.values.toList() },
|
||||
) { fromMqtt, fromAir ->
|
||||
// Expiry has to be driven by a clock, not by arrivals. Both upstream flows only re-emit
|
||||
// when a DENM arrives, so a sender that simply stops transmitting - drives away, loses
|
||||
// power, leaves range - would otherwise leave its hazard on the map forever: there is no
|
||||
// further emission to recompute the list. This tick is what makes a hazard fade.
|
||||
tickerFlow(DENM_EXPIRY_TICK_MS),
|
||||
) { fromMqtt, fromAir, _ ->
|
||||
val now = System.currentTimeMillis()
|
||||
(fromMqtt + fromAir)
|
||||
.filterNot { it.isTermination } // the hazard is over - stop drawing it
|
||||
.associateBy { it.dedupKey } // last write wins = most recent per hazard
|
||||
.values
|
||||
// Not heard from in DENM_TTL_MS: treat as gone. DENMs repeat at roughly 1 Hz, so a
|
||||
// full minute of silence is ~60 missed repetitions - well past "we briefly lost one".
|
||||
.filter { now - it.timestamp <= DENM_TTL_MS }
|
||||
.sortedByDescending { it.timestamp }
|
||||
}.stateIn(viewModelScope, SharingStarted.Eagerly, emptyList())
|
||||
|
||||
/**
|
||||
* Live signal state per intersection, newest first, keyed by [IntersectionSignalState.key].
|
||||
*
|
||||
* ESP32-C5 path only: SPATEM arrives over the air on BTP port 2004. The CiT One path publishes
|
||||
* SPATEM on its own MQTT topic in a different (protobuf-wrapped) shape, which is not wired up.
|
||||
*
|
||||
* One entry per intersection, not per message: SPATEM repeats at ~2 Hz per RSU, so a log would
|
||||
* grow without telling anyone anything. Entries expire like DENMs do - an intersection left
|
||||
* behind stops transmitting, and the same clock-driven argument applies.
|
||||
*/
|
||||
val spatIntersections: StateFlow<List<SpatIntersection>> = combine(
|
||||
camUseCaseRepository.airSpat
|
||||
.runningFold(emptyMap<String, SpatIntersection>()) { acc, spat ->
|
||||
acc + spat.intersections.associate { i ->
|
||||
i.key to SpatIntersection(i, spat.stationId, spat.rssiDbm, spat.timestamp)
|
||||
}
|
||||
},
|
||||
tickerFlow(SPAT_EXPIRY_TICK_MS),
|
||||
) { byKey, _ ->
|
||||
val now = System.currentTimeMillis()
|
||||
byKey.values
|
||||
.filter { now - it.timestamp <= SPAT_TTL_MS }
|
||||
.sortedByDescending { it.timestamp }
|
||||
}.stateIn(viewModelScope, SharingStarted.Eagerly, emptyList())
|
||||
|
||||
/** Emits immediately, then every [periodMs], purely to re-trigger a time-dependent combine. */
|
||||
private fun tickerFlow(periodMs: Long): Flow<Long> = flow {
|
||||
while (true) {
|
||||
emit(System.currentTimeMillis())
|
||||
delay(periodMs)
|
||||
}
|
||||
}
|
||||
|
||||
private companion object {
|
||||
/** Use Case API topic carrying received DENMs (CiT One path only). */
|
||||
const val DENM_RX_TOPIC = "v2x-uca/output/json/denm"
|
||||
|
||||
/**
|
||||
* How long a hazard stays listed after its last repetition. A DENM has no "still here"
|
||||
* guarantee beyond the sender repeating it, and its own validityDuration is not decoded
|
||||
* yet, so silence is the only expiry signal available.
|
||||
*/
|
||||
const val DENM_TTL_MS = 60_000L
|
||||
|
||||
/** How often the list is re-evaluated for expiry. Sets the worst-case lateness of a fade. */
|
||||
const val DENM_EXPIRY_TICK_MS = 5_000L
|
||||
|
||||
/**
|
||||
* SPATEM repeats at ~2 Hz, so 15 s of silence is ~30 missed repetitions: the RSU is out of
|
||||
* range. Much shorter than the DENM window because a stale traffic light is more
|
||||
* misleading than a stale hazard - a light that stopped updating is not "still green".
|
||||
*/
|
||||
const val SPAT_TTL_MS = 15_000L
|
||||
const val SPAT_EXPIRY_TICK_MS = 2_000L
|
||||
}
|
||||
|
||||
// ── DENM transmission ─────────────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user