NEW: distance-matrix-to-pd-series-transformer

Author: cherman2

q2_types/distance_matrix/_deferred_setup/_transformers.py

@@ -7,6 +7,7 @@
 # ----------------------------------------------------------------------------
 
 import skbio
+import pandas as pd
 
 from .. import LSMatFormat
 
@@ -24,3 +25,34 @@ def _1(data: skbio.DistanceMatrix) -> LSMatFormat:
 @plugin.register_transformer
 def _2(ff: LSMatFormat) -> skbio.DistanceMatrix:
     return skbio.DistanceMatrix.read(str(ff), format='lsmat', verify=False)
+
+
+@plugin.register_transformer
+def _3(ff: LSMatFormat) -> pd.Series:
+    dm = skbio.DistanceMatrix.read(str(ff), format='lsmat', verify=False)
+    series = dm.to_series()
+    assert series.size != 0, ("Distance Matrix must contain more"
+                              "than one sample")
+    return series
+
+
+@plugin.register_transformer
+def _4(data: pd.Series) -> skbio.DistanceMatrix:
+    ids = data.index.get_level_values(0).unique().union(
+         data.index.get_level_values(1).unique(), sort=False).values
+    dm_df = pd.DataFrame(data=[], index=ids, columns=ids)
+    for index, row in dm_df.iterrows():
+        dm_df.loc[index, index] = float(0)
+        for col in dm_df.columns:
+            if dm_df.loc[index, col] != 0:
+                try:
+                    dm_df.loc[index, col] = data[index, col]
+                    dm_df.loc[col, index] = data[index, col]
+                except KeyError:
+                    dm_df.loc[index, col] = data[col, index]
+                    dm_df.loc[col, index] = data[col, index]
+    dm = skbio.DistanceMatrix(dm_df, ids=dm_df.index)
+    ff = LSMatFormat()
+    with ff.open() as fh:
+        dm.write(fh, format='lsmat')
+    return ff

q2_types/distance_matrix/tests/test_transformers.py

@@ -9,6 +9,7 @@
 import unittest
 
 import skbio
+import pandas as pd
 
 from q2_types.distance_matrix import LSMatFormat
 from qiime2.plugin.testing import TestPluginBase
@@ -40,6 +41,36 @@ def test_lsmat_format_to_skbio_distance_matrix(self):
             exp = skbio.DistanceMatrix.read(str(input))
             self.assertEqual(obs, exp)
 
+    def test_lsmat_format_to_pd_series(self):
+        filenames = ('distance-matrix-2x2.tsv',
+                     'distance-matrix-NxN.tsv')
+        for filename in filenames:
+            input, obs = self.transform_format(LSMatFormat,
+                                               pd.Series,
+                                               filename=filename)
+            exp = skbio.DistanceMatrix.read(str(input)).to_series()
+        pd.testing.assert_series_equal(obs, exp)
+
+    def test_lsmat_format_to_pd_series_1x1(self):
+        filename = 'distance-matrix-1x1.tsv'
+        with self.assertRaisesRegex(AssertionError, "Distance Matrix *"):
+            self.transform_format(LSMatFormat,
+                                  pd.Series,
+                                  filename=filename)
+
+    def test_pd_series_to_skbio_distance_matrix(self):
+        transformer = self.get_transformer(pd.Series, skbio.DistanceMatrix)
+
+        filenames = ('distance-matrix-NxN.tsv', 'distance-matrix-2x2.tsv')
+        for filename in filenames:
+            input = skbio.DistanceMatrix.read(self.get_data_path(filename))
+            obs = transformer(input.to_series())
+            obs = skbio.DistanceMatrix.read(str(obs))
+
+            exp = input
+
+            self.assertEqual(obs, exp)
+
 
 if __name__ == "__main__":
     unittest.main()