ENH: add core score action

q2_feature_table/__init__.py

@@ -20,6 +20,7 @@
 from ._group import group
 from ._rename import rename_ids
 from ._heatmap import (heatmap, heatmap_choices)
+from ._core_score import core_score
 
 try:
     from ._version import __version__
@@ -33,4 +34,4 @@
            'filter_seqs', 'subsample_ids', 'rename_ids',
            'filter_features_conditionally', 'split',
            'tabulate_feature_frequencies', 'tabulate_sample_frequencies',
-           'summarize', 'normalize']
+           'summarize', 'normalize', 'core_score']

q2_feature_table/_core_score.py

@@ -0,0 +1,39 @@
+# ----------------------------------------------------------------------------
+# Copyright (c) 2016-2026, QIIME 2 development team.
+#
+# Distributed under the terms of the Modified BSD License.
+#
+# The full license is in the file LICENSE, distributed with this software.
+# ----------------------------------------------------------------------------
+
+import numpy as np
+import pandas as pd
+
+
+def core_score(
+        table: pd.DataFrame,
+        min_rel_abundance: float = 1e-3,
+        mean_abundance_on_presence: bool = False,
+        eps: float = 1e-6
+) -> pd.DataFrame:
+    """Compute prevalence-abundance core scores for feature table columns."""
+    prevalence = (table > min_rel_abundance).mean(axis=0)
+    if mean_abundance_on_presence:
+        mean_abundance = table.where(
+            table > min_rel_abundance).mean(axis=0).fillna(0)
+    else:
+        mean_abundance = table.mean(axis=0)
+    log_mean = np.log10(mean_abundance + eps)
+
+    prevalence_scaled = _minmax_scale(prevalence, eps)
+    log_mean_scaled = _minmax_scale(log_mean, eps)
+    score = prevalence_scaled * log_mean_scaled
+
+    result = pd.DataFrame({'core_score': score})
+    result.index.name = 'id'
+    return result
+
+
+def _minmax_scale(x, eps):
+    """Scale values to the range [0, 1] with an offset denominator."""
+    return (x - x.min()) / (x.max() - x.min() + eps)

q2_feature_table/plugin_setup.py

@@ -14,7 +14,8 @@
 from q2_types.feature_table import (
     FeatureTable, Frequency, RelativeFrequency, PresenceAbsence, Composition)
 from q2_types.feature_data import (
-    FeatureData, Sequence, Taxonomy, AlignedSequence, SequenceCharacteristics)
+    FeatureData, Sequence, Taxonomy, AlignedSequence, SequenceCharacteristics,
+    Importance)
 from q2_types.metadata import ImmutableMetadata
 
 import q2_feature_table
@@ -776,3 +777,44 @@
     "composition with common methods for RNA-seq.",
     citations=[citations["Zmrzlikar_RNAnorm_RNA-seq_data_2023"]],
 )
+
+plugin.methods.register_function(
+    function=q2_feature_table.core_score,
+    inputs={'table': FeatureTable[RelativeFrequency]},
+    parameters={
+        'min_rel_abundance': Float % Range(0, 1, inclusive_start=False),
+        'mean_abundance_on_presence': Bool,
+        'eps': Float % Range(0, None, inclusive_start=False)
+    },
+    outputs=[('core_scores', FeatureData[Importance])],
+    input_descriptions={
+        'table': 'The relative-frequency feature table to score.'
+    },
+    parameter_descriptions={
+        'min_rel_abundance': (
+            'Minimum relative abundance used to count a feature as present '
+            'in a sample.'
+        ),
+        'mean_abundance_on_presence': (
+            'If true, compute mean relative abundance using only samples '
+            'where the feature is greater than `min_rel_abundance`. If '
+            'false, compute mean relative abundance across all samples.'
+        ),
+        'eps': (
+            'Small positive value used as the log offset and min-max '
+            'scaling denominator offset.'
+        )
+    },
+    output_descriptions={
+        'core_scores': 'Core scores for each feature.'
+    },
+    name='Compute core score for features',
+    description=('Compute a prevalence-abundance core score for each feature '
+                 'in a relative-frequency feature table. Prevalence is the '
+                 'fraction of samples where a feature exceeds '
+                 '`min_rel_abundance`; mean relative abundance is computed '
+                 'either across all samples or only samples above '
+                 '`min_rel_abundance`, then log-transformed with `eps`; both '
+                 'quantities are min-max scaled across features and '
+                 'multiplied to produce the final score.')
+)

q2_feature_table/tests/test_core_score.py

@@ -0,0 +1,93 @@
+# ----------------------------------------------------------------------------
+# Copyright (c) 2016-2026, QIIME 2 development team.
+#
+# Distributed under the terms of the Modified BSD License.
+#
+# The full license is in the file LICENSE, distributed with this software.
+# ----------------------------------------------------------------------------
+
+from unittest import main
+
+import numpy as np
+import pandas as pd
+import pandas.testing as pdt
+from rachis.plugin.testing import TestPluginBase
+
+from q2_feature_table import core_score
+from q2_feature_table._core_score import _minmax_scale
+
+
+class TestCoreScore(TestPluginBase):
+    package = 'q2_feature_table.tests'
+
+    def setUp(self):
+        super().setUp()
+        self.table = pd.DataFrame(
+            np.array([
+                [0.0, 0.005, 0.0],
+                [0.02, 0.005, 0.0],
+                [0.04, 0.005, 0.3],
+            ]),
+            index=['S1', 'S2', 'S3'],
+            columns=['O1', 'O2', 'O3']
+        )
+
+    def test_core_score(self):
+        observed = core_score(self.table)
+
+        prevalence = np.array([2 / 3, 1, 1 / 3])
+        mean_abundance = np.array([0.06 / 3, 0.005, 0.1])
+        log_mean = np.log10(mean_abundance + 1e-6)
+        exp_prevalence = _minmax_scale(prevalence, 1e-6)
+        exp_log_mean = _minmax_scale(log_mean, 1e-6)
+        expected = pd.DataFrame(
+            {'core_score': exp_prevalence * exp_log_mean},
+            index=pd.Index(['O1', 'O2', 'O3'], name='id')
+        )
+
+        pdt.assert_frame_equal(observed, expected)
+
+    def test_core_score_values_are_finite(self):
+        observed = core_score(self.table)
+        self.assertTrue(np.isfinite(observed.to_numpy()).all())
+
+    def test_core_score_custom_parameters(self):
+        observed = core_score(
+            self.table, min_rel_abundance=0.01, eps=1e-3)
+
+        prevalence = np.array([2 / 3, 0, 1 / 3])
+        mean_abundance = np.array([0.06 / 3, 0.005, 0.1])
+        log_mean = np.log10(mean_abundance + 1e-3)
+        exp_prevalence = _minmax_scale(prevalence, 1e-3)
+        exp_log_mean = _minmax_scale(log_mean, 1e-3)
+        expected = pd.DataFrame(
+            {'core_score': exp_prevalence * exp_log_mean},
+            index=pd.Index(['O1', 'O2', 'O3'], name='id')
+        )
+
+        pdt.assert_frame_equal(observed, expected)
+
+    def test_core_score_mean_abundance_on_presence(self):
+        observed = core_score(
+            self.table, min_rel_abundance=0.01,
+            mean_abundance_on_presence=True)
+
+        prevalence = np.array([2 / 3, 0, 1 / 3])
+        mean_abundance = np.array([0.03, 0, 0.3])
+        log_mean = np.log10(mean_abundance + 1e-6)
+        exp_prevalence = _minmax_scale(prevalence, 1e-6)
+        exp_log_mean = _minmax_scale(log_mean, 1e-6)
+        expected = pd.DataFrame(
+            {'core_score': exp_prevalence * exp_log_mean},
+            index=pd.Index(['O1', 'O2', 'O3'], name='id')
+        )
+
+        pdt.assert_frame_equal(observed, expected)
+
+    def test_minmax_scale(self):
+        observed = _minmax_scale(pd.Series([2, 2, 2]), eps=1e-6)
+        self.assertTrue(np.isfinite(observed.to_numpy()).all())
+
+
+if __name__ == "__main__":
+    main()