closes #45 Created new mc method clip_recovery. 0.4.8

CHANGELOG.rst

@@ -1,3 +1,12 @@
+Geomet 0.4.8 (2024-11-03)
+=========================
+
+Feature
+-------
+
+- Created new mc method: clip_recovery (#45)
+
+
 Geomet 0.4.7 (2024-10-31)
 =========================
 

elphick/geomet/base.py

@@ -4,10 +4,12 @@
 import re
 from abc import ABC
 from pathlib import Path
-from typing import Optional, Union, Literal, TypeVar, TYPE_CHECKING, Any
+from typing import Optional, Union, Literal, TypeVar, TYPE_CHECKING
 
 import numpy as np
 import pandas as pd
+import plotly.express as px
+import plotly.graph_objects as go
 
 from elphick.geomet.config import read_yaml
 from elphick.geomet.utils.components import get_components, is_compositional
@@ -17,8 +19,6 @@
 from elphick.geomet.utils.timer import log_timer
 from .config.config_read import get_column_config
 from .plot import parallel_plot, comparison_plot
-import plotly.express as px
-import plotly.graph_objects as go
 
 if TYPE_CHECKING:
     from elphick.geomet.flowsheet.stream import Stream
@@ -138,6 +138,22 @@ def mass_data(self, value):
         # Recalculate the aggregate whenever the data changes
         self.aggregate = self.weight_average()
 
+    def get_mass_data(self, include_moisture: bool = True) -> pd.DataFrame:
+        """Get the mass data
+
+        Args:
+            include_moisture: If True (and moisture is in scope), include the moisture mass column
+
+        Returns:
+
+        """
+        if include_moisture and self.moisture_in_scope:
+            moisture_mass = self._mass_data[self.mass_wet_var] - self._mass_data[self.mass_dry_var]
+            mass_data: pd.DataFrame = self._mass_data.copy()
+            mass_data.insert(loc=2, column=self.moisture_column, value=moisture_mass)
+            return mass_data
+        return self._mass_data
+
     @property
     def aggregate(self) -> pd.DataFrame:
         if self._aggregate is None and self._mass_data is not None:
@@ -235,6 +251,49 @@ def balance_composition(self) -> MC:
 
         return self
 
+    def clip_recovery(self, other: MC, recovery_bounds: tuple[float, float] = (0.01, 0.99),
+                      allow_moisture_coercion: bool = True) -> MC:
+        """Clip the recovery to the specified bounds and recalculate the estimate.
+
+        Args:
+            other: The other MassComposition object, from which the recovery of self is calculated.
+            recovery_bounds: The bounds for the recovery between 0.0 and 1.0
+            allow_moisture_coercion: if True, allow the wet mass to be modified to maintain the moisture (in the
+            case that dry mass is clipped to manage recovery)
+
+        Returns:
+            The MassComposition object with the recovery clipped to the bounds.
+        """
+        recovery: pd.DataFrame = (self.get_mass_data(include_moisture=False) /
+                                  other.get_mass_data(include_moisture=False))
+
+        # Limit the recovery to the bounds
+        before_clip = recovery.copy()
+        recovery = recovery.clip(lower=recovery_bounds[0], upper=recovery_bounds[1]).fillna(0.0)
+
+        # Check if any records were affected
+        affected_indexes = set(recovery.index[np.any(before_clip != recovery, axis=1)])
+        if affected_indexes:
+            # Recalculate the estimate from the bound recovery
+            new_mass: pd.DataFrame = recovery * other.get_mass_data(include_moisture=False)[recovery.columns]
+
+            if self.moisture_in_scope and allow_moisture_coercion:
+                # Calculate the moisture from the new mass
+                new_mass[self.mass_wet_var] = solve_mass_moisture(mass_dry=new_mass[self.mass_dry_var],
+                                                                  moisture=self.data[self.moisture_column])
+
+            # Log the top 50 records affected by the recovery coercion
+            affected_indexes_list = sorted(affected_indexes)[:50]
+            self._logger.info(f"Recovery coercion affected {len(affected_indexes)} records. "
+                              f"Affected indexes (first 50): {affected_indexes_list}")
+
+            # Update the mass data of self
+            self.update_mass_data(new_mass)
+        else:
+            self._logger.info("Recovery coercion did not affect any records.")
+
+        return self
+
     def set_moisture(self, moisture: Union[pd.Series, float, int], mass_to_adjust: Literal['wet', 'dry'] = 'wet') -> MC:
         """Set the moisture to the specified value
 

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "geometallurgy"
 packages = [{ include = "elphick/geomet" }]
-version = "0.4.7"
+version = "0.4.8"
 description = "Tools for the geometallurgist"
 authors = ["Greg <[email protected]>"]
 repository = "https://github.com/elphick/geometallurgy"

tests/test_017_clip_recovery.py

@@ -0,0 +1,78 @@
+import pandas as pd
+import pytest
+
+from elphick.geomet import Sample
+from elphick.geomet.base import MassComposition
+
+
+@pytest.fixture
+def sample_data():
+    data = {
+        'mass_wet': [100., 200., 300.],
+        'mass_dry': [90., 180., 270.],
+        'component_1': [10., 20., 30.],
+        'component_2': [5., 10., 15.]
+    }
+    return pd.DataFrame(data)
+
+
+@pytest.fixture
+def excess_recovery_sample_data():
+    data = {
+        'mass_wet': [100., 200., 300.],
+        'mass_dry': [90., 180., 290.],
+        'component_1': [10., 20., 40.],
+        'component_2': [5., 10., 15.]
+    }
+    return pd.DataFrame(data)
+
+
+@pytest.fixture
+def other_data():
+    data = {
+        'mass_wet': [110., 210., 310.],
+        'mass_dry': [100., 190., 280.],
+        'component_1': [12., 22., 32.],
+        'component_2': [6., 11., 16.]
+    }
+    return pd.DataFrame(data)
+
+
+def test_clip_recovery_no_clip(sample_data, other_data):
+    mc = Sample(data=sample_data, name='sample', mass_wet_var='mass_wet', mass_dry_var='mass_dry',
+                component_vars=['component_1', 'component_2'])
+    other_mc = Sample(data=other_data, name='other', mass_wet_var='mass_wet', mass_dry_var='mass_dry',
+                      component_vars=['component_1', 'component_2'])
+
+    recovery_bounds = (0.01, 0.99)
+    result = mc.clip_recovery(other_mc, recovery_bounds)
+
+    assert isinstance(result, MassComposition)
+    assert not result._mass_data.empty
+    assert 'mass_wet' in result._mass_data.columns
+    assert 'mass_dry' in result._mass_data.columns
+    assert 'component_1' in result._mass_data.columns
+    assert 'component_2' in result._mass_data.columns
+
+    # no changes were made, since the recovery was within bounds
+    pd.testing.assert_frame_equal(result.data.drop(columns='H2O'), sample_data.drop(columns='h2o'))
+
+
+def test_clip_recovery_clipped(excess_recovery_sample_data, other_data):
+    mc = Sample(data=excess_recovery_sample_data, name='sample', mass_wet_var='mass_wet', mass_dry_var='mass_dry',
+                component_vars=['component_1', 'component_2'])
+    original_moisture: pd.Series = mc.data['H2O']
+    other_mc = Sample(data=other_data, name='other', mass_wet_var='mass_wet', mass_dry_var='mass_dry',
+                      component_vars=['component_1', 'component_2'])
+
+    recovery_bounds = (0.01, 0.99)
+    result = mc.clip_recovery(other_mc, recovery_bounds)
+
+    expected_result: pd.DataFrame = pd.DataFrame(
+        {'mass_wet': {0: 100.0, 1: 200.0, 2: 286.75862}, 'mass_dry': {0: 90.0, 1: 180.0, 2: 277.2},
+         'H2O': {0: 10.0, 1: 10.0, 2: 3.333333}, 'component_1': {0: 10.0, 1: 20.0, 2: 32.0},
+         'component_2': {0: 5.0, 1: 10.0, 2: 15.692640692640694}})
+
+    pd.testing.assert_frame_equal(result.data, expected_result)
+
+    pd.testing.assert_series_equal(result.data['H2O'], original_moisture)