Repository: deepmodeling/dpgen2 (language: python)

Issue:
Implement the op CollectData
This op 
1. collect the labeled data stored in each fp task directories,
2. store the data in one directory
3. add the directory to the training data.

The source code of the op is [here](https://github.com/deepmodeling/dpgen2/blob/master/dpgen2/op/collect_data.py). One may consult the docstr for the interface of the op.


Discussion / hints:

Fix patch:
diff --git a/dpgen2/op/collect_data.py b/dpgen2/op/collect_data.py
index f5f255b9..79b93cd2 100644
--- a/dpgen2/op/collect_data.py
+++ b/dpgen2/op/collect_data.py
@@ -4,7 +4,7 @@
     OPIOSign,
     Artifact
 )
-import os, json
+import os, json, dpdata
 from typing import Tuple, List, Set
 from pathlib import Path
 
@@ -38,7 +38,9 @@ def execute(
             self,
             ip : OPIO,
     ) -> OPIO:
-        r"""Execute the OP.
+        r"""Execute the OP. This OP collect data scattered in directories given by `ip['labeled_data']` 
+        in to one `dpdata.Multisystems` and store it in a directory named `name`. This directory is appended
+        to the list `iter_data`.
 
         Parameters
         ----------
@@ -46,15 +48,30 @@ def execute(
             Input dict with components:
         
             - `name`: (`str`) The name of this iteration. The data generated by this iteration will be place in a sub-directory of `name`.
-            - `labeled_data`: (`Artifact(List[Path])`) The paths of labeled data generated by FP tasks.
-            - `iter_data`: (`Artifact(set[Path])`) The paths of iteration data.
+            - `labeled_data`: (`Artifact(List[Path])`) The paths of labeled data generated by FP tasks of the current iteration.
+            - `iter_data`: (`Artifact(List[Path])`) The data paths previous iterations.
 
         Returns
         -------
             Output dict with components:
         
-            - `iter_data`: (`Artifact(List[Path])`) The paths of iteration data, added with labeled data generated by this iteration.
+            - `iter_data`: (`Artifact(List[Path])`) The data paths of previous and the current iteration data.
         
         """
-        raise NotImplementedError
+        name = ip['name']
+        labeled_data = ip['labeled_data']
+        iter_data = ip['iter_data']
+
+        ms = dpdata.MultiSystems()
+        for ii in labeled_data:
+            ss = dpdata.LabeledSystem(ii, fmt='deepmd/npy')
+            ms.append(ss)
+        
+        ms.to_deepmd_npy(name)
+        iter_data.append(Path(name))
+
+        return OPIO({
+            'iter_data' : iter_data,
+        })
+            
 


Test patch:
diff --git a/tests/op/test_collect_data.py b/tests/op/test_collect_data.py
new file mode 100644
index 00000000..a0d8c2c9
--- /dev/null
+++ b/tests/op/test_collect_data.py
@@ -0,0 +1,69 @@
+import dpdata
+from op.context import dpgen2
+import numpy as np
+import unittest, json, shutil
+from mock import mock, patch, call
+from dflow.python import (
+    OP,
+    OPIO,
+    OPIOSign,
+    Artifact,
+    TransientError,
+)
+from pathlib import Path
+from dpgen2.constants import (
+    lmp_conf_name,
+    lmp_input_name,
+    lmp_log_name,
+    lmp_traj_name,
+    lmp_model_devi_name,
+)
+from dpgen2.op.collect_data import CollectData
+from fake_data_set import fake_system, fake_multi_sys
+
+class TestRunLmp(unittest.TestCase):
+    def setUp(self):
+        self.iter_data = [Path('iter0')]
+        self.labeled_data = [Path('data0'), Path('data1')]    
+        # fake multi sys
+        self.atom_name_ms = 'foo'
+        self.natoms = [1, 2]
+        self.nframes = [3, 4]
+        ms = fake_multi_sys(self.nframes, self.natoms, self.atom_name_ms)
+        # fake sys
+        self.atom_name = 'bar'
+        self.natoms_0 = 3
+        self.natoms_1 = 4
+        self.nframes_0 = 6
+        self.nframes_1 = 5
+        ss_0 = fake_system(self.nframes_0, self.natoms_0, self.atom_name)
+        ss_1 = fake_system(self.nframes_1, self.natoms_1, self.atom_name)        
+        # dump
+        ms.to_deepmd_npy(self.iter_data[0])
+        ss_0.to_deepmd_npy(self.labeled_data[0])
+        ss_1.to_deepmd_npy(self.labeled_data[1])
+
+    def tearDown(self):
+        for ii in ['iter0', 'iter1']:
+            if Path(ii).is_dir():
+                shutil.rmtree(ii)        
+        for ii in ['data0', 'data1']:
+            if Path(ii).is_dir():
+                shutil.rmtree(ii)        
+
+    def test_success(self):
+        op = CollectData()
+        out = op.execute(
+            OPIO({
+                'name' : 'iter1',
+                'iter_data' : self.iter_data,
+                'labeled_data' : self.labeled_data,
+            }))
+        self.assertTrue(Path('iter1').is_dir())
+        self.assertEqual(out['iter_data'], [Path('iter0'), Path('iter1')])
+        ms = dpdata.MultiSystems()
+        ms.from_deepmd_npy(out['iter_data'][1])
+        self.assertEqual(sorted(ms.systems.keys()), ['bar3', 'bar4'])
+        self.assertEqual(ms.systems['bar3'].get_nframes(), 6)
+        self.assertEqual(ms.systems['bar4'].get_nframes(), 5)
+