///|
/// Output of sensor quality marking before event analysis.
pub(all) struct SensorQualityResult {
readings : Array[Reading]
health : Array[SensorHealth]
diagnostics : Array[Diagnostic]
} derive(Debug, Eq)
///|
/// Mark a contiguous index range with one quality flag.
fn flag_range(
readings : Array[Reading],
start : Int,
end_exclusive : Int,
flag : QualityFlag,
) -> Unit {
let bounded_start = clamp_int(start, 0, readings.length())
let bounded_end = clamp_int(end_exclusive, bounded_start, readings.length())
for index = bounded_start; index < bounded_end; index = index + 1 {
readings[index] = readings[index].add_flag(flag)
}
}
///|
/// Detect long runs where the measured value barely changes.
fn mark_stuck_runs(readings : Array[Reading], profile : SensorProfile) -> Int {
if readings.length() < profile.stuck_minimum_samples {
return 0
}
let mut run_start = 0
let mut runs = 0
for index = 1; index <= readings.length(); index = index + 1 {
let continues = if index < readings.length() {
abs_double(
readings[index].temperature_c - readings[index - 1].temperature_c,
) <=
profile.stuck_tolerance_c
} else {
false
}
if !continues {
let run_length = index - run_start
if run_length >= profile.stuck_minimum_samples {
flag_range(readings, run_start, index, SensorStuck)
runs = runs + 1
}
run_start = index
}
}
runs
}
///|
/// Mark abrupt sample-to-sample temperature changes.
fn mark_noisy_steps(readings : Array[Reading], profile : SensorProfile) -> Int {
let mut count = 0
for index = 1; index < readings.length(); index = index + 1 {
let step = abs_double(
readings[index].temperature_c - readings[index - 1].temperature_c,
)
if step >= profile.noise_step_c {
readings[index] = readings[index].add_flag(SensorNoisy)
readings[index - 1] = readings[index - 1].add_flag(SensorNoisy)
count = count + 1
}
}
count
}
///|
/// True when values are monotonic in either direction over a window.
fn monotonic_direction(
readings : Array[Reading],
start : Int,
end_exclusive : Int,
) -> Int {
let mut increasing = true
let mut decreasing = true
for index = start + 1; index < end_exclusive; index = index + 1 {
if readings[index].temperature_c < readings[index - 1].temperature_c {
increasing = false
}
if readings[index].temperature_c > readings[index - 1].temperature_c {
decreasing = false
}
}
if increasing {
1
} else if decreasing {
-1
} else {
0
}
}
///|
/// Detect sustained monotonic drift windows.
fn mark_drift_windows(
readings : Array[Reading],
profile : SensorProfile,
) -> Int {
let width = profile.drift_window_samples
if width < 2 || readings.length() < width {
return 0
}
let mut windows = 0
let mut start = 0
while start + width <= readings.length() {
let end_exclusive = start + width
let change = readings[end_exclusive - 1].temperature_c -
readings[start].temperature_c
if monotonic_direction(readings, start, end_exclusive) != 0 &&
abs_double(change) >= profile.drift_threshold_c {
flag_range(readings, start, end_exclusive, SensorDrifting)
windows = windows + 1
start = end_exclusive
} else {
start = start + 1
}
}
windows
}
///|
/// Mark explicit device status failures.
fn mark_offline_status(readings : Array[Reading]) -> Int {
let mut count = 0
for index = 0; index < readings.length(); index = index + 1 {
let status = readings[index].status.trim().to_owned().to_lower()
if status == "offline" ||
status == "disconnected" ||
status == "error" ||
status == "failed" {
readings[index] = readings[index].add_flag(SensorOffline)
count = count + 1
}
}
count
}
///|
/// Count samples with battery below the configurable reporting threshold.
fn count_low_battery(readings : Array[Reading], threshold : Double) -> Int {
let mut count = 0
for sample in readings {
match sample.battery_percent {
Some(value) => if value <= threshold { count = count + 1 }
None => ()
}
}
count
}
///|
/// Count readings containing at least one sensor health flag.
fn count_health_flags(readings : Array[Reading]) -> Int {
let mut count = 0
for sample in readings {
if sample.has_flag(SensorStuck) ||
sample.has_flag(SensorNoisy) ||
sample.has_flag(SensorDrifting) ||
sample.has_flag(SensorOffline) ||
sample.has_flag(SensorConflict) {
count = count + 1
}
}
count
}
///|
/// Compare all sensors sharing a timestamp and mark disagreements.
pub fn mark_sensor_conflicts(
input : Array[Reading],
threshold_c : Double,
) -> ReadingBatch {
let readings = sort_readings(input)
let diagnostics : Array[Diagnostic] = []
let by_time : Map[Int64, Array[Int]] = Map([])
for index = 0; index < readings.length(); index = index + 1 {
let timestamp = readings[index].timestamp
match by_time.get(timestamp) {
Some(indices) => indices.push(index)
None => by_time[timestamp] = [index]
}
}
for timestamp in by_time.keys() {
let indices = by_time[timestamp]
if indices.length() < 2 {
continue
}
let mut minimum = readings[indices[0]].temperature_c
let mut maximum = minimum
for index in indices {
let value = readings[index].temperature_c
if value < minimum {
minimum = value
}
if value > maximum {
maximum = value
}
}
if maximum - minimum >= threshold_c {
for index in indices {
readings[index] = readings[index].add_flag(SensorConflict)
}
diagnostics.push(
warning_diagnostic(
"sensor.cross_sensor_conflict",
"sensors at the same timestamp disagree by \{maximum - minimum} °C",
),
)
}
}
{ readings, diagnostics }
}
///|
/// Count gaps belonging to one sensor.
fn count_sensor_gaps(gaps : Array[SamplingGap], sensor_id : String) -> Int {
let mut count = 0
for gap in gaps {
if gap.sensor_id == sensor_id {
count = count + 1
}
}
count
}
///|
/// Count conflict flags in one sensor stream.
fn count_conflicts(readings : Array[Reading]) -> Int {
let mut count = 0
for sample in readings {
if sample.has_flag(SensorConflict) {
count = count + 1
}
}
count
}
///|
/// Convert issue counts to a bounded sensor health score.
fn health_score(
stuck_runs : Int,
noisy_steps : Int,
drift_windows : Int,
offline_samples : Int,
gaps : Int,
conflicts : Int,
low_battery_samples : Int,
) -> Int {
clamp_int(
100 -
stuck_runs * 15 -
noisy_steps * 4 -
drift_windows * 8 -
offline_samples * 10 -
gaps * 8 -
conflicts * 3 -
low_battery_samples,
0,
100,
)
}
///|
/// Build concise observations for one health summary.
fn health_observations(
stuck_runs : Int,
noisy_steps : Int,
drift_windows : Int,
offline_samples : Int,
gaps : Int,
conflicts : Int,
low_battery_samples : Int,
) -> Array[String] {
let observations : Array[String] = []
if stuck_runs > 0 {
observations.push("\{stuck_runs} stuck-value runs detected")
}
if noisy_steps > 0 {
observations.push("\{noisy_steps} abrupt temperature steps detected")
}
if drift_windows > 0 {
observations.push("\{drift_windows} monotonic drift windows detected")
}
if offline_samples > 0 {
observations.push("\{offline_samples} explicit offline samples detected")
}
if gaps > 0 {
observations.push("\{gaps} sampling gaps detected")
}
if conflicts > 0 {
observations.push("\{conflicts} cross-sensor conflicts detected")
}
if low_battery_samples > 0 {
observations.push("\{low_battery_samples} low-battery samples detected")
}
if observations.length() == 0 {
observations.push("no sensor health anomalies detected")
}
observations
}
///|
/// Mark within-sensor quality anomalies and calculate health summaries.
pub fn analyze_sensor_quality(
input : Array[Reading],
gaps : Array[SamplingGap],
config : AnalysisConfig,
) -> SensorQualityResult {
let conflict_result = mark_sensor_conflicts(
input,
config.conflict_threshold_c,
)
let diagnostics = copy_array(conflict_result.diagnostics)
let output : Array[Reading] = []
let health : Array[SensorHealth] = []
for sensor_id in unique_sensor_ids(conflict_result.readings) {
let stream = sort_readings(
readings_for_sensor(conflict_result.readings, sensor_id),
)
let profile = config.profile_for(sensor_id)
let stuck_runs = mark_stuck_runs(stream, profile)
let noisy_steps = mark_noisy_steps(stream, profile)
let drift_windows = mark_drift_windows(stream, profile)
let offline_samples = mark_offline_status(stream)
let sensor_gaps = count_sensor_gaps(gaps, sensor_id)
let conflicts = count_conflicts(stream)
let low_battery_samples = count_low_battery(stream, 20.0)
let flagged_samples = count_health_flags(stream)
let score = health_score(
stuck_runs, noisy_steps, drift_windows, offline_samples, sensor_gaps, conflicts,
low_battery_samples,
)
if score < 100 {
diagnostics.push(
warning_diagnostic(
"sensor.health_degraded",
"sensor health score is \{score}",
).for_sensor(sensor_id),
)
}
health.push({
sensor_id,
score,
stuck_runs,
noisy_steps,
drift_windows,
gaps: sensor_gaps,
conflicts,
low_battery_samples,
flagged_samples,
observations: health_observations(
stuck_runs, noisy_steps, drift_windows, offline_samples, sensor_gaps, conflicts,
low_battery_samples,
),
})
output.append(stream)
}
{ readings: sort_readings(output), health, diagnostics }
}
///|
/// Find the weakest sensor health summary.
pub fn weakest_sensor(health : Array[SensorHealth]) -> SensorHealth? {
if health.length() == 0 {
return None
}
let mut weakest = health[0]
for item in health {
if item.score < weakest.score {
weakest = item
}
}
Some(weakest)
}
///|
/// Average sensor health score.
pub fn average_sensor_health(health : Array[SensorHealth]) -> Int? {
if health.length() == 0 {
return None
}
let mut total = 0
for item in health {
total = total + item.score
}
Some(total / health.length())
}