test_spss.py 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166
  1. import datetime
  2. from pathlib import Path
  3. import numpy as np
  4. import pytest
  5. import pandas as pd
  6. import pandas._testing as tm
  7. from pandas.util.version import Version
  8. pyreadstat = pytest.importorskip("pyreadstat")
  9. # TODO(CoW) - detection of chained assignment in cython
  10. # https://github.com/pandas-dev/pandas/issues/51315
  11. @pytest.mark.filterwarnings("ignore::pandas.errors.ChainedAssignmentError")
  12. @pytest.mark.filterwarnings("ignore:ChainedAssignmentError:FutureWarning")
  13. @pytest.mark.parametrize("path_klass", [lambda p: p, Path])
  14. def test_spss_labelled_num(path_klass, datapath):
  15. # test file from the Haven project (https://haven.tidyverse.org/)
  16. # Licence at LICENSES/HAVEN_LICENSE, LICENSES/HAVEN_MIT
  17. fname = path_klass(datapath("io", "data", "spss", "labelled-num.sav"))
  18. df = pd.read_spss(fname, convert_categoricals=True)
  19. expected = pd.DataFrame({"VAR00002": "This is one"}, index=[0])
  20. expected["VAR00002"] = pd.Categorical(expected["VAR00002"])
  21. tm.assert_frame_equal(df, expected)
  22. df = pd.read_spss(fname, convert_categoricals=False)
  23. expected = pd.DataFrame({"VAR00002": 1.0}, index=[0])
  24. tm.assert_frame_equal(df, expected)
  25. @pytest.mark.filterwarnings("ignore::pandas.errors.ChainedAssignmentError")
  26. @pytest.mark.filterwarnings("ignore:ChainedAssignmentError:FutureWarning")
  27. def test_spss_labelled_num_na(datapath):
  28. # test file from the Haven project (https://haven.tidyverse.org/)
  29. # Licence at LICENSES/HAVEN_LICENSE, LICENSES/HAVEN_MIT
  30. fname = datapath("io", "data", "spss", "labelled-num-na.sav")
  31. df = pd.read_spss(fname, convert_categoricals=True)
  32. expected = pd.DataFrame({"VAR00002": ["This is one", None]})
  33. expected["VAR00002"] = pd.Categorical(expected["VAR00002"])
  34. tm.assert_frame_equal(df, expected)
  35. df = pd.read_spss(fname, convert_categoricals=False)
  36. expected = pd.DataFrame({"VAR00002": [1.0, np.nan]})
  37. tm.assert_frame_equal(df, expected)
  38. @pytest.mark.filterwarnings("ignore::pandas.errors.ChainedAssignmentError")
  39. @pytest.mark.filterwarnings("ignore:ChainedAssignmentError:FutureWarning")
  40. def test_spss_labelled_str(datapath):
  41. # test file from the Haven project (https://haven.tidyverse.org/)
  42. # Licence at LICENSES/HAVEN_LICENSE, LICENSES/HAVEN_MIT
  43. fname = datapath("io", "data", "spss", "labelled-str.sav")
  44. df = pd.read_spss(fname, convert_categoricals=True)
  45. expected = pd.DataFrame({"gender": ["Male", "Female"]})
  46. expected["gender"] = pd.Categorical(expected["gender"])
  47. tm.assert_frame_equal(df, expected)
  48. df = pd.read_spss(fname, convert_categoricals=False)
  49. expected = pd.DataFrame({"gender": ["M", "F"]})
  50. tm.assert_frame_equal(df, expected)
  51. @pytest.mark.filterwarnings("ignore::pandas.errors.ChainedAssignmentError")
  52. @pytest.mark.filterwarnings("ignore:ChainedAssignmentError:FutureWarning")
  53. def test_spss_umlauts(datapath):
  54. # test file from the Haven project (https://haven.tidyverse.org/)
  55. # Licence at LICENSES/HAVEN_LICENSE, LICENSES/HAVEN_MIT
  56. fname = datapath("io", "data", "spss", "umlauts.sav")
  57. df = pd.read_spss(fname, convert_categoricals=True)
  58. expected = pd.DataFrame(
  59. {"var1": ["the ä umlaut", "the ü umlaut", "the ä umlaut", "the ö umlaut"]}
  60. )
  61. expected["var1"] = pd.Categorical(expected["var1"])
  62. tm.assert_frame_equal(df, expected)
  63. df = pd.read_spss(fname, convert_categoricals=False)
  64. expected = pd.DataFrame({"var1": [1.0, 2.0, 1.0, 3.0]})
  65. tm.assert_frame_equal(df, expected)
  66. def test_spss_usecols(datapath):
  67. # usecols must be list-like
  68. fname = datapath("io", "data", "spss", "labelled-num.sav")
  69. with pytest.raises(TypeError, match="usecols must be list-like."):
  70. pd.read_spss(fname, usecols="VAR00002")
  71. def test_spss_umlauts_dtype_backend(datapath, dtype_backend):
  72. # test file from the Haven project (https://haven.tidyverse.org/)
  73. # Licence at LICENSES/HAVEN_LICENSE, LICENSES/HAVEN_MIT
  74. fname = datapath("io", "data", "spss", "umlauts.sav")
  75. df = pd.read_spss(fname, convert_categoricals=False, dtype_backend=dtype_backend)
  76. expected = pd.DataFrame({"var1": [1.0, 2.0, 1.0, 3.0]}, dtype="Int64")
  77. if dtype_backend == "pyarrow":
  78. pa = pytest.importorskip("pyarrow")
  79. from pandas.arrays import ArrowExtensionArray
  80. expected = pd.DataFrame(
  81. {
  82. col: ArrowExtensionArray(pa.array(expected[col], from_pandas=True))
  83. for col in expected.columns
  84. }
  85. )
  86. tm.assert_frame_equal(df, expected)
  87. def test_invalid_dtype_backend():
  88. msg = (
  89. "dtype_backend numpy is invalid, only 'numpy_nullable' and "
  90. "'pyarrow' are allowed."
  91. )
  92. with pytest.raises(ValueError, match=msg):
  93. pd.read_spss("test", dtype_backend="numpy")
  94. @pytest.mark.filterwarnings("ignore::pandas.errors.ChainedAssignmentError")
  95. @pytest.mark.filterwarnings("ignore:ChainedAssignmentError:FutureWarning")
  96. def test_spss_metadata(datapath):
  97. # GH 54264
  98. fname = datapath("io", "data", "spss", "labelled-num.sav")
  99. df = pd.read_spss(fname)
  100. metadata = {
  101. "column_names": ["VAR00002"],
  102. "column_labels": [None],
  103. "column_names_to_labels": {"VAR00002": None},
  104. "file_encoding": "UTF-8",
  105. "number_columns": 1,
  106. "number_rows": 1,
  107. "variable_value_labels": {"VAR00002": {1.0: "This is one"}},
  108. "value_labels": {"labels0": {1.0: "This is one"}},
  109. "variable_to_label": {"VAR00002": "labels0"},
  110. "notes": [],
  111. "original_variable_types": {"VAR00002": "F8.0"},
  112. "readstat_variable_types": {"VAR00002": "double"},
  113. "table_name": None,
  114. "missing_ranges": {},
  115. "missing_user_values": {},
  116. "variable_storage_width": {"VAR00002": 8},
  117. "variable_display_width": {"VAR00002": 8},
  118. "variable_alignment": {"VAR00002": "unknown"},
  119. "variable_measure": {"VAR00002": "unknown"},
  120. "file_label": None,
  121. "file_format": "sav/zsav",
  122. }
  123. if Version(pyreadstat.__version__) >= Version("1.2.4"):
  124. metadata.update(
  125. {
  126. "creation_time": datetime.datetime(2015, 2, 6, 14, 33, 36),
  127. "modification_time": datetime.datetime(2015, 2, 6, 14, 33, 36),
  128. }
  129. )
  130. if Version(pyreadstat.__version__) >= Version("1.2.8"):
  131. metadata["mr_sets"] = {}
  132. tm.assert_dict_equal(df.attrs, metadata)