Automatic push from FunctionsAPI

Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+  edition = "2024"
+  name    = "web"
+  version = "0.1.0"
+
+[dependencies]
+  fathom-function      = { git = "ssh://git@github.com/fathom-io/pipeline-calculations.git", branch = "main" }
+  pipeline-application = { git = "ssh://git@github.com/fathom-io/pipeline-calculations.git", branch = "main" }
+  serde                = { version = "1.0.219", features = ["derive"] }
+  tokio                = { version = "1.43.0", features = ["macros", "rt-multi-thread"] }
+  uom                  = { version = "0.36" }
+  uuid                 = { version = "1" }

README.md

@@ -1,2 +1,63 @@
-# 1e6cf0434f894be0b862fdba1b17f652
+# ILI comparison (compare all)
 
+This function will run all possible ILI comparisons for a given pipeline using the continuous scorer implementation.
+
+See [ILI comparison](../ili_comparison/README.md) for more details of the comparison algorithm.
+
+## Input
+
+### Arguments
+
+- `org_id`: as string which should be a valid `uuid` for the organization
+- `project_id`: the id of the data project where the pipeline data is found
+- `pipeline_id`: a `array` of string values representing a valid `uuid` for a pipeline
+- `weld_location_threshold`: a `float` value
+- `feature_location_threshold`: a `float` value
+- `upstream_girth_threshold`: a `float` value
+- `orientation_threshold`: a `float` value
+- `anomaly_size`: an `object` with the configuration for the depth and length of the anomalies
+- `surface_location_criteria`: a `string` whose value should be one of
+  - `matching`
+  - `any`
+
+## Creating the function on the platform
+
+To create this function on the platform using the `cli` set up the port forwarding as shown in README.
+
+Then run the following command to create the function.
+
+```bash
+cargo run functions create \
+  -f functions/ili_compare_all_continuous \
+  -d "Runs the ILI comparison algorithm for all combinations of ILI reports for the given pipeline using the continuous scorer" \
+  -o org_id=string \
+  -o project_id=string \
+  -o pipeline_ids=array \
+  -o unique_pipeline_ids=array \
+  -o matched_ids=array \
+  -o unmatched_ids=array \
+  -o summary_ids=array \
+  -i org_id=string \
+  -i project_id=string \
+  -i pipeline_id=array \
+  -i weld_location_threshold=float \
+  -i upstream_girth_threshold=float \
+  -i feature_location_threshold=float \
+  -i orientation_threshold=float \
+  -i anomaly_size=object \
+  -i surface_location_criteria=string 
+```
+
+## Testing the function locally
+
+You can run and test the function locally by running
+
+```bash
+cargo run
+```
+
+Then you can check it work with `curl` as follows
+
+```bash
+curl localhost:8080 -d $(jq '. | tojson' functions/ili_compare_all_continuous/example_input.json)
+```

example_input.json

@@ -0,0 +1,52 @@
+{
+  "anomaly_size": {
+    "depth": {
+      "center": {
+        "inputType": "number",
+        "value": "5.0"
+      },
+      "left_width": {
+        "inputType": "number",
+        "value": "15.1"
+      },
+      "right_width": {
+        "inputType": "number",
+        "value": "25.1"
+      }
+    },
+    "length": {
+      "center": {
+        "inputType": "number",
+        "value": "5.0"
+      },
+      "left_width": {
+        "inputType": "number",
+        "value": "60.1"
+      },
+      "right_width": {
+        "inputType": "number",
+        "value": "10.1"
+      }
+    }
+  },
+  "feature_location_threshold": "1",
+  "org_id": "b4d6cbfd-c444-4a0b-9fcd-39984c68e860",
+  "orientation_threshold": "20",
+  "pipelineId_dataProductId": "019759fa-c251-7c92-96c3-ce9100b2f403",
+  "pipelineId_organizationId": "b4d6cbfd-c444-4a0b-9fcd-39984c68e860",
+  "pipelineId_resourceClientIds": [
+    "pipeline:1"
+  ],
+  "pipelineId_resourceIds": [
+    "019759fb-52a6-7db3-86ed-ed4505af4837"
+  ],
+  "pipelineId_resourceTypeId": 1,
+  "pipelineId_workspaceId": "68482a0d64a45c363e1bea9e",
+  "pipeline_id": [
+    "019759fb-52a6-7db3-86ed-ed4505af4837"
+  ],
+  "project_id": "68482a0d64a45c363e1bea9e",
+  "surface_location_criteria": "matching",
+  "upstream_girth_threshold": "0.101",
+  "weld_location_threshold": "1"
+}

src/main.rs

@@ -0,0 +1,161 @@
+use fathom_function::{forms::TableCellValue, tracing};
+use pipeline_application::{
+    application::{
+        Application, AsF64, ContinuousScoreMatcher, IliComparisonOutput,
+        SurfaceLocationCriteria as SLC, TwoSidedContinuousScorer,
+    },
+    serialization::{serialize_meter, serialize_orientation_min},
+    units::{frequency::percent_per_year, velocity::millimeter_per_year},
+};
+use uom::si::f64::{Angle, Frequency, Length, Velocity};
+use uuid::Uuid;
+
+#[fathom_function::function]
+async fn ili_compare_all(input: Input) -> Result<Output, String> {
+    let mut output = Output::new(input.org_id, &input.project_id);
+    let app = Application::new_from_compile_env(input.org_id, &input.project_id).unwrap();
+
+    for pipeline_id in input.pipeline_id {
+        let result = app
+            .ili_compare_all(pipeline_id, ContinuousScoreMatcher::from(&input.criteria))
+            .await
+            .map_err(|err| {
+                tracing::error!(%pipeline_id, ?err, "Error running comparison algorithm");
+                format!("{err:?}")
+            })?;
+        output.push_result(pipeline_id, result);
+    }
+
+    Ok(output)
+}
+
+#[derive(Debug, serde::Serialize, Default)]
+struct Output {
+    // TODO: remove org_id, project_id, unique_pipeline_ids once function outputs from previous
+    // nodes are all visible in downstream nodes (FTHM-11016).
+    org_id: Uuid,
+    unique_pipeline_ids: Vec<Uuid>,
+    project_id: String,
+    pipeline_ids: Vec<Uuid>,
+    matched_ids: Vec<Uuid>,
+    unmatched_ids: Vec<Uuid>,
+    summary_ids: Vec<Uuid>,
+}
+
+impl Output {
+    fn new(org_id: Uuid, project_id: impl ToString) -> Self {
+        Self {
+            org_id,
+            project_id: project_id.to_string(),
+            ..Default::default()
+        }
+    }
+
+    fn push_result(&mut self, pipeline_id: Uuid, ili_comparison_result: Vec<IliComparisonOutput>) {
+        self.unique_pipeline_ids.push(pipeline_id);
+        ili_comparison_result.into_iter().for_each(|res| {
+            self.pipeline_ids.push(pipeline_id);
+            self.matched_ids.push(res.matched_id);
+            self.unmatched_ids.push(res.unmatched_id);
+            self.summary_ids.push(res.summary_id);
+        });
+    }
+}
+
+#[derive(Debug, serde::Deserialize)]
+struct Input {
+    org_id: Uuid,
+    project_id: String,
+    pipeline_id: Vec<Uuid>,
+
+    #[serde(flatten)]
+    criteria: Criteria,
+}
+
+#[derive(Debug, serde::Deserialize)]
+struct Criteria {
+    #[serde(with = "serialize_meter")]
+    weld_location_threshold: Length,
+    #[serde(with = "serialize_meter")]
+    feature_location_threshold: Length,
+    #[serde(with = "serialize_meter")]
+    upstream_girth_threshold: Length,
+    #[serde(with = "serialize_orientation_min")]
+    orientation_threshold: Angle,
+    anomaly_size: AnomalySizeCriteria,
+    surface_location_criteria: SurfaceLocationCriteria,
+}
+
+#[derive(Debug, serde::Deserialize)]
+struct AnomalySizeCriteria {
+    depth: SizeCriteria<TableCellValue>,
+    length: SizeCriteria<TableCellValue>,
+}
+
+#[derive(Debug, serde::Deserialize)]
+struct SizeCriteria<T> {
+    left_width: T,
+    center: T,
+    right_width: T,
+}
+
+impl SizeCriteria<TableCellValue> {
+    fn transform<T>(&self, f: impl Fn(f64) -> T) -> Option<SizeCriteria<T>> {
+        Some(SizeCriteria {
+            left_width: Option::<f64>::try_from(&self.left_width).ok()?.map(&f)?,
+            center: Option::<f64>::try_from(&self.center).ok()?.map(&f)?,
+            right_width: Option::<f64>::try_from(&self.right_width).ok()?.map(&f)?,
+        })
+    }
+}
+
+impl<T> From<SizeCriteria<T>> for TwoSidedContinuousScorer<T>
+where
+    T: AsF64 + Default,
+{
+    fn from(value: SizeCriteria<T>) -> Self {
+        TwoSidedContinuousScorer::default()
+            .with_left_width(value.left_width)
+            .with_center(value.center)
+            .with_right_width(value.right_width)
+    }
+}
+
+#[derive(Debug, Clone, Copy, serde::Deserialize)]
+#[serde(rename_all = "snake_case")]
+enum SurfaceLocationCriteria {
+    Matching,
+    Any,
+}
+
+impl From<SurfaceLocationCriteria> for SLC {
+    fn from(value: SurfaceLocationCriteria) -> Self {
+        match value {
+            SurfaceLocationCriteria::Matching => Self::Matching,
+            SurfaceLocationCriteria::Any => Self::Any,
+        }
+    }
+}
+
+impl From<&Criteria> for ContinuousScoreMatcher {
+    fn from(value: &Criteria) -> Self {
+        let length = value
+            .anomaly_size
+            .length
+            .transform(Velocity::new::<millimeter_per_year>)
+            .map(TwoSidedContinuousScorer::from);
+        let depth = value
+            .anomaly_size
+            .depth
+            .transform(Frequency::new::<percent_per_year>)
+            .map(TwoSidedContinuousScorer::from);
+        ContinuousScoreMatcher::default()
+            .with_weld_location_threshold(value.weld_location_threshold)
+            .with_feature_location_threshold(value.feature_location_threshold)
+            .with_upstream_girth_threshold(value.upstream_girth_threshold)
+            .with_orientation_threshold(value.orientation_threshold)
+            .with_depth_scorer(depth)
+            .with_length_scorer(length)
+            .with_surface_location_criteria(value.surface_location_criteria)
+    }
+}