///|
pub struct DataQuality {
  name : String
  completeness : Float
  validity : Float
  timeliness : Float
  consistency : Float
} derive(Debug, Eq)

///|
pub fn data_quality(
  name : String,
  completeness : Float,
  validity : Float,
  timeliness : Float,
  consistency : Float,
) -> DataQuality {
  { name, completeness, validity, timeliness, consistency }
}

///|
pub fn DataQuality::score(self : DataQuality) -> Float {
  (self.completeness + self.validity + self.timeliness + self.consistency) / 4.0
}

///|
pub fn DataQuality::is_valid(self : DataQuality) -> Bool {
  self.completeness >= 0.0 &&
  self.completeness <= 1.0 &&
  self.validity >= 0.0 &&
  self.validity <= 1.0 &&
  self.timeliness >= 0.0 &&
  self.timeliness <= 1.0 &&
  self.consistency >= 0.0 &&
  self.consistency <= 1.0
}

///|
pub fn DataQuality::is_acceptable(self : DataQuality, minimum : Float) -> Bool {
  self.is_valid() && self.score() >= minimum
}

///|
pub struct QualityDimension {
  name : String
  passed : Int
  total : Int
} derive(Debug, Eq)

///|
pub fn quality_dimension(
  name : String,
  passed : Int,
  total : Int,
) -> QualityDimension {
  { name, passed, total }
}

///|
pub fn QualityDimension::ratio(self : QualityDimension) -> Float {
  if self.total <= 0 {
    0.0
  } else {
    Float::from_int(self.passed) / Float::from_int(self.total)
  }
}

///|
pub fn QualityDimension::is_acceptable(
  self : QualityDimension,
  minimum : Float,
) -> Bool {
  self.passed >= 0 && self.total >= self.passed && self.ratio() >= minimum
}

///|
pub struct QualityProfile {
  name : String
  dimensions : Array[QualityDimension]
} derive(Debug, Eq)

///|
pub fn quality_profile(
  name : String,
  dimensions : Array[QualityDimension],
) -> QualityProfile {
  { name, dimensions }
}

///|
pub fn QualityProfile::score(self : QualityProfile) -> Float {
  if self.dimensions.length() == 0 {
    0.0
  } else {
    summarize(self.dimensions.map(fn(item) { item.ratio() })).mean
  }
}

///|
pub fn QualityProfile::is_acceptable(
  self : QualityProfile,
  minimum : Float,
) -> Bool {
  for dimension in self.dimensions {
    if !dimension.is_acceptable(minimum) {
      return false
    }
  }
  true
}

///|
pub fn QualityProfile::weakest(self : QualityProfile) -> QualityDimension? {
  if self.dimensions.length() == 0 {
    None
  } else {
    let mut result = self.dimensions[0]
    for dimension in self.dimensions[1:] {
      if dimension.ratio() < result.ratio() {
        result = dimension
      }
    }
    Some(result)
  }
}

///|
pub fn QualityProfile::to_table(self : QualityProfile) -> ReportTable {
  let rows : Array[Array[String]] = []
  for dimension in self.dimensions {
    rows.push([
      dimension.name,
      "{dimension.passed}",
      "{dimension.total}",
      "{dimension.ratio()}",
    ])
  }
  table(["dimension", "passed", "total", "ratio"], rows)
}

///|
pub enum MissingPolicy {
  RejectMissing
  IgnoreMissing
  UseDefault
} derive(Debug, Eq)

///|
pub fn reject_missing() -> MissingPolicy {
  RejectMissing
}

///|
pub fn ignore_missing() -> MissingPolicy {
  IgnoreMissing
}

///|
pub fn use_default() -> MissingPolicy {
  UseDefault
}

///|
pub struct DataField {
  name : String
  value : String?
  expected_unit : String?
  policy : MissingPolicy
} derive(Debug, Eq)

///|
pub fn data_field(
  name : String,
  value : String?,
  expected_unit : String?,
  policy : MissingPolicy,
) -> DataField {
  { name, value, expected_unit, policy }
}

///|
pub fn DataField::is_missing(self : DataField) -> Bool {
  self.value is None
}

///|
pub fn DataField::is_acceptable(self : DataField) -> Bool {
  match self.value {
    Some(value) => value.trim().length() > 0
    None => self.policy is IgnoreMissing || self.policy is UseDefault
  }
}

///|
pub fn validate_fields(fields : Array[DataField]) -> Array[String] {
  let result : Array[String] = []
  for field in fields {
    if !field.is_acceptable() {
      result.push(field.name + " is missing or empty")
    }
  }
  result
}

///|
pub fn completeness(fields : Array[DataField]) -> Float {
  if fields.length() == 0 {
    1.0
  } else {
    Float::from_int(fields.length() - validate_fields(fields).length()) /
    Float::from_int(fields.length())
  }
}

///|
pub fn data_quality_gate(fields : Array[DataField], minimum : Float) -> Bool {
  completeness(fields) >= minimum
}

///|
pub fn profile_from_fields(
  name : String,
  fields : Array[DataField],
) -> QualityProfile {
  let total = fields.length()
  let passed = total - validate_fields(fields).length()
  quality_profile(name, [quality_dimension("completeness", passed, total)])
}

///|
pub fn quality_report(profiles : Array[QualityProfile]) -> ReportTable {
  let rows : Array[Array[String]] = []
  for profile in profiles {
    let weakest_name = match profile.weakest() {
      Some(value) => value.name
      None => "none"
    }
    rows.push([profile.name, "\{profile.score()}", weakest_name])
  }
  table(["profile", "score", "weakest"], rows)
}