Skip to content

omnipy.components.pandas.serializers

CLASS DESCRIPTION
PandasDatasetToTarFileSerializer

PandasDatasetToTarFileSerializer

Bases: TarFileSerializer[PandasDataset]

METHOD DESCRIPTION
deserialize
get_dataset_cls_for_new
get_output_file_suffix
is_dataset_directly_supported
serialize
Source code in src/omnipy/components/pandas/serializers.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class PandasDatasetToTarFileSerializer(TarFileSerializer[PandasDataset]):
    """"""
    @classmethod
    def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
        return isinstance(dataset, PandasDataset)

    @classmethod
    def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
        return PandasDataset

    @classmethod
    def get_output_file_suffix(cls) -> str:
        return 'csv'

    @classmethod
    def serialize(cls, dataset: PandasDataset) -> bytes | memoryview:
        assert isinstance(dataset, PandasDataset)

        def pandas_encode_func(pandas_data: 'pd.DataFrame') -> memoryview:
            csv_bytes = BytesIO()
            pandas_data.to_csv(csv_bytes, encoding='utf8', mode='wb', index=False)
            return csv_bytes.getbuffer()

        return cls.create_tarfile_from_dataset(dataset, data_encode_func=pandas_encode_func)

    @classmethod
    def deserialize(cls, serialized: bytes, any_file_suffix=False) -> PandasDataset:
        pandas_dataset = PandasDataset()

        def csv_decode_func(file_stream: IO[bytes]) -> 'pd.DataFrame':
            from .lazy_import import pd
            return pd.read_csv(file_stream, encoding='utf8')

        def python_dictify_object(data_file: str, obj_val: Any) -> dict:
            return {data_file: obj_val}

        cls.create_dataset_from_tarfile(
            pandas_dataset,
            serialized,
            data_decode_func=csv_decode_func,
            dictify_object_func=python_dictify_object,
            import_method='from_data',
            any_file_suffix=any_file_suffix,
        )  # noqa

        return pandas_dataset

deserialize classmethod 

deserialize(serialized: bytes, any_file_suffix=False) -> PandasDataset
Source code in src/omnipy/components/pandas/serializers.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@classmethod
def deserialize(cls, serialized: bytes, any_file_suffix=False) -> PandasDataset:
    pandas_dataset = PandasDataset()

    def csv_decode_func(file_stream: IO[bytes]) -> 'pd.DataFrame':
        from .lazy_import import pd
        return pd.read_csv(file_stream, encoding='utf8')

    def python_dictify_object(data_file: str, obj_val: Any) -> dict:
        return {data_file: obj_val}

    cls.create_dataset_from_tarfile(
        pandas_dataset,
        serialized,
        data_decode_func=csv_decode_func,
        dictify_object_func=python_dictify_object,
        import_method='from_data',
        any_file_suffix=any_file_suffix,
    )  # noqa

    return pandas_dataset

get_dataset_cls_for_new classmethod 

get_dataset_cls_for_new() -> Type[IsDataset]
Source code in src/omnipy/components/pandas/serializers.py 
20
21
22
@classmethod
def get_dataset_cls_for_new(cls) -> Type[IsDataset]:
    return PandasDataset

get_output_file_suffix classmethod 

get_output_file_suffix() -> str
Source code in src/omnipy/components/pandas/serializers.py 
24
25
26
@classmethod
def get_output_file_suffix(cls) -> str:
    return 'csv'

is_dataset_directly_supported classmethod 

is_dataset_directly_supported(dataset: IsDataset) -> bool
Source code in src/omnipy/components/pandas/serializers.py 
16
17
18
@classmethod
def is_dataset_directly_supported(cls, dataset: IsDataset) -> bool:
    return isinstance(dataset, PandasDataset)

serialize classmethod 

serialize(dataset: PandasDataset) -> bytes | memoryview
Source code in src/omnipy/components/pandas/serializers.py 
28
29
30
31
32
33
34
35
36
37
@classmethod
def serialize(cls, dataset: PandasDataset) -> bytes | memoryview:
    assert isinstance(dataset, PandasDataset)

    def pandas_encode_func(pandas_data: 'pd.DataFrame') -> memoryview:
        csv_bytes = BytesIO()
        pandas_data.to_csv(csv_bytes, encoding='utf8', mode='wb', index=False)
        return csv_bytes.getbuffer()

    return cls.create_tarfile_from_dataset(dataset, data_encode_func=pandas_encode_func)