chore(context): improve io handler

Signed-off-by: weiwee <wbwmat@gmail.com>

python/fate/arch/context/_context.py

@@ -9,9 +9,9 @@
 from ..unify import device
 from ._cipher import CipherKit
 from ._federation import GC, Parties, Party
-from ._io import IOKit
 from ._namespace import Namespace
 from ._tensor import TensorKit
+from .io.kit import IOKit
 from .metric import MetricsWrap
 
 

python/fate/arch/context/io/data/csv.py

@@ -0,0 +1,38 @@
+from ....unify import URI
+from .df import Dataframe
+
+
+class CSVReader:
+    def __init__(self, ctx, name: str, uri: URI, metadata: dict) -> None:
+        self.name = name
+        self.ctx = ctx
+        self.uri = uri
+        self.metadata = metadata
+
+    def read_dataframe(self):
+        import inspect
+
+        from fate.arch import dataframe
+
+        kwargs = {}
+        p = inspect.signature(dataframe.CSVReader.__init__).parameters
+        parameter_keys = p.keys()
+        for k, v in self.metadata.items():
+            if k in parameter_keys:
+                kwargs[k] = v
+
+        dataframe_reader = dataframe.CSVReader(**kwargs).to_frame(self.ctx, self.uri.path)
+        # s_df = dataframe.serialize(self.ctx, dataframe_reader)
+        # dataframe_reader = dataframe.deserialize(self.ctx, s_df)
+        return Dataframe(dataframe_reader, dataframe_reader.shape[1], dataframe_reader.shape[0])
+
+
+class CSVWriter:
+    def __init__(self, ctx, name: str, uri: URI, metadata: dict) -> None:
+        self.name = name
+        self.ctx = ctx
+        self.uri = uri
+        self.metadata = metadata
+
+    def write_dataframe(self, df):
+        ...

python/fate/arch/context/io/data/dataframe.py

@@ -0,0 +1,38 @@
+from ....unify import URI
+from .df import Dataframe
+
+
+class DataFrameWriter:
+    def __init__(self, ctx, name: str, uri, metadata: dict) -> None:
+        self.name = name
+        self.ctx = ctx
+        self.uri = uri
+        self.metadata = metadata
+
+    def write_dataframe(self, df):
+        import json
+
+        from fate.arch import dataframe
+
+        with open(self.uri, "w") as f:
+            data_dict = dataframe.serialize(self.ctx, df)
+            json.dump(data_dict, f)
+
+
+class DataFrameReader:
+    def __init__(self, ctx, name: str, uri: URI, metadata: dict) -> None:
+        self.name = name
+        self.ctx = ctx
+        self.uri = uri
+        self.metadata = metadata
+
+    def read_dataframe(self):
+        import json
+
+        from fate.arch import dataframe
+
+        with open(self.uri.path, "r") as fin:
+            data_dict = json.loads(fin.read())
+            df = dataframe.deserialize(self.ctx, data_dict)
+
+        return Dataframe(df, df.shape[1], df.shape[0])

python/fate/arch/context/io/data/df.py

@@ -0,0 +1,11 @@
+class Dataframe:
+    def __init__(self, frames, num_features, num_samples) -> None:
+        self.data = frames
+        self.num_features = num_features
+        self.num_samples = num_samples
+
+    def __len__(self):
+        return self.num_samples
+
+    def to_local(self):
+        return self.data.to_local()