from __future__ import annotations
import abc
import warnings
from collections import ChainMap
from collections.abc import Callable, Mapping, MutableMapping, Sequence
from dataclasses import dataclass
from fractions import Fraction
from types import MappingProxyType, SimpleNamespace
from typing import Any, Final, NamedTuple, Union, cast
import numpy as np
import pandas as pd
import xarray
from typing_extensions import override
from .. import parser
ConversionFunction = Callable[[Union[parser.RData, parser.RObject]], Any]
[docs]
class RLanguage(NamedTuple):
"""R language construct."""
elements: list[Any]
attributes: Mapping[str, Any]
[docs]
class RExpression(NamedTuple):
"""R expression."""
elements: list[RLanguage]
[docs]
@dataclass
class RBuiltin:
"""R builtin."""
name: str
[docs]
@dataclass
class RFunction:
"""R function."""
environment: Mapping[str, Any]
formals: Mapping[str, Any] | None
body: RLanguage
attributes: Mapping[str, Any]
@property
def source(self) -> str:
return "\n".join(self.attributes["srcref"].srcfile.lines)
[docs]
@dataclass
class RExternalPointer:
"""R bytecode."""
protected: Any
tag: Any
[docs]
@dataclass
class RBytecode:
"""R bytecode."""
code: xarray.DataArray
constants: Sequence[Any]
attributes: Mapping[str, Any]
[docs]
class REnvironment(ChainMap[str, Any]):
"""R environment."""
def __init__(
self,
*maps: MutableMapping[str, Any],
frame: Mapping[str, Any] | None = None,
) -> None:
super().__init__(*maps)
self.frame = frame
def convert_list(
r_list: parser.RObject,
conversion_function: ConversionFunction,
) -> Mapping[str, Any] | list[Any]:
"""
Expand a tagged R pairlist to a Python dictionary.
Args:
r_list: Pairlist R object, with tags.
conversion_function: Conversion function to apply to the elements of
the list. By default is the identity function.
Returns:
A dictionary with the tags of the pairwise list as keys and their
corresponding values as values.
See Also:
convert_vector
"""
if r_list.info.type is parser.RObjectType.NILVALUE:
return {}
if r_list.info.type not in {
parser.RObjectType.LIST,
parser.RObjectType.LANG,
}:
msg = "Must receive a LIST, LANG or NILVALUE object"
raise TypeError(msg)
tag = None if r_list.tag is None else conversion_function(r_list.tag)
cdr = conversion_function(r_list.value[1])
if tag is not None:
if cdr is None:
cdr = {}
return {tag: conversion_function(r_list.value[0]), **cdr}
if cdr is None:
cdr = []
return [conversion_function(r_list.value[0]), *cdr]
def convert_env(
r_env: parser.RObject,
conversion_function: ConversionFunction,
) -> REnvironment:
"""Convert environment objects."""
if r_env.info.type is not parser.RObjectType.ENV:
msg = "Must receive a ENV object"
raise TypeError(msg)
frame = conversion_function(r_env.value.frame)
enclosure = conversion_function(r_env.value.enclosure)
hash_table = conversion_function(r_env.value.hash_table)
dictionary = {}
if hash_table is not None:
for d in hash_table:
if d is not None:
dictionary.update(d)
return REnvironment(dictionary, enclosure, frame=frame)
def convert_attrs(
r_obj: parser.RObject,
conversion_function: ConversionFunction,
) -> Mapping[str, Any]:
"""
Return the attributes of an object as a Python dictionary.
Args:
r_obj: R object.
conversion_function: Conversion function to apply to the elements of
the attribute list. By default is the identity function.
Returns:
A dictionary with the names of the attributes as keys and their
corresponding values as values.
See Also:
convert_list
"""
if r_obj.attributes:
attrs = cast(
Mapping[str, Any],
conversion_function(r_obj.attributes),
)
else:
attrs = {}
return attrs
def convert_vector(
r_vec: parser.RObject,
conversion_function: ConversionFunction,
attrs: Mapping[str, Any] | None = None,
) -> list[Any] | Mapping[str, Any]:
"""
Convert a R vector to a Python list or dictionary.
If the vector has a ``names`` attribute, the result is a dictionary with
the names as keys. Otherwise, the result is a Python list.
Args:
r_vec: R vector.
conversion_function: Conversion function to apply to the elements of
the vector. By default is the identity function.
attrs: Attributes of the vector.
Returns:
A dictionary with the ``names`` of the vector as keys and their
corresponding values as values. If the vector does not have an
argument ``names``, then a normal Python list is returned.
See Also:
convert_list
"""
if attrs is None:
attrs = {}
if r_vec.info.type not in {
parser.RObjectType.VEC,
parser.RObjectType.EXPR,
}:
msg = "Must receive a VEC or EXPR object"
raise TypeError(msg)
value: list[Any] | Mapping[str, Any] = [
conversion_function(o) for o in r_vec.value
]
# If it has the name attribute, use a dict instead
field_names = attrs.get("names")
if field_names is not None:
value = dict(zip(field_names, value))
return value
def safe_decode(byte_str: bytes, encoding: str) -> str | bytes:
"""Decode a (possibly malformed) string."""
try:
return byte_str.decode(encoding)
except UnicodeDecodeError as e:
warnings.warn( # noqa: B028
f"Exception while decoding {byte_str!r}: {e}",
)
return byte_str
def convert_char(
r_char: parser.RObject,
*,
default_encoding: str | None = None,
force_default_encoding: bool = False,
) -> str | bytes | None:
"""
Decode a R character array to a Python string or bytes.
The bits that signal the encoding are in the general pointer. The
string can be encoded in UTF8, LATIN1 or ASCII, or can be a sequence
of bytes.
Args:
r_char: R character array.
default_encoding: Default encoding to apply when encoding info
is not available.
force_default_encoding: Always use the default encoding.
Returns:
Decoded string.
See Also:
convert_symbol
"""
if r_char.info.type is not parser.RObjectType.CHAR:
msg = "Must receive a CHAR object"
raise TypeError(msg)
if r_char.value is None:
return None
assert isinstance(r_char.value, bytes)
encoding = None
if not force_default_encoding:
if r_char.info.gp & parser.CharFlags.UTF8:
encoding = "utf_8"
elif r_char.info.gp & parser.CharFlags.LATIN1:
encoding = "latin_1"
elif r_char.info.gp & parser.CharFlags.ASCII:
encoding = "ascii"
elif r_char.info.gp & parser.CharFlags.BYTES:
encoding = "bytes"
if encoding is None:
if default_encoding:
encoding = default_encoding
else:
# Assume ASCII if no encoding is marked
warnings.warn("Unknown encoding. Assumed ASCII.") # noqa: B028
encoding = "ascii"
return (
r_char.value
if encoding == "bytes"
else safe_decode(r_char.value, encoding)
)
def convert_symbol(
r_symbol: parser.RObject,
conversion_function: ConversionFunction,
) -> str | bytes:
"""
Decode a R symbol to a Python string or bytes.
Args:
r_symbol: R symbol.
conversion_function: Conversion function to apply to the char element
of the symbol. By default is the identity function.
Returns:
Decoded string.
See Also:
convert_char
"""
if r_symbol.info.type is parser.RObjectType.SYM:
symbol = conversion_function(r_symbol.value)
assert isinstance(symbol, str)
return symbol
msg = "Must receive a SYM object"
raise TypeError(msg)
def convert_array(
r_array: parser.RObject,
attrs: Mapping[str, Any] | None = None,
) -> np.ndarray[Any, Any] | xarray.DataArray:
"""
Convert a R array to a Numpy ndarray or a Xarray DataArray.
If the array has attribute ``dimnames`` the output will be a
Xarray DataArray, preserving the dimension names.
Args:
r_array: R array.
attrs: Attributes of the array.
Returns:
Array.
See Also:
convert_vector
"""
if attrs is None:
attrs = {}
if r_array.info.type not in {
parser.RObjectType.LGL,
parser.RObjectType.INT,
parser.RObjectType.REAL,
parser.RObjectType.CPLX,
}:
msg = "Must receive an array object"
raise TypeError(msg)
value = r_array.value
shape = attrs.get("dim")
if shape is not None:
# R matrix order is like FORTRAN
value = np.reshape(value, shape, order="F")
dimension_names = None
coords = None
dimnames = attrs.get("dimnames")
if dimnames:
if isinstance(dimnames, Mapping):
dimension_names = list(dimnames.keys())
coords = dimnames
else:
dimension_names = [f"dim_{i}" for i, _ in enumerate(dimnames)]
coords = {
dimension_names[i]: d
for i, d in enumerate(dimnames)
if d is not None
}
value = xarray.DataArray(
value,
dims=dimension_names,
coords=coords,
)
return value # type: ignore [no-any-return]
R_INT_MIN = -2**31
def _dataframe_column_transform(source: Any) -> Any: # noqa: ANN401
if isinstance(source, np.ndarray):
if np.issubdtype(source.dtype, np.integer):
return pd.Series(source, dtype=pd.Int32Dtype()).array
if np.issubdtype(source.dtype, np.bool_):
return pd.Series(source, dtype=pd.BooleanDtype()).array
if np.issubdtype(source.dtype, np.str_):
return pd.Series(source, dtype=pd.StringDtype()).array
return source
def dataframe_constructor(
obj: Mapping[str, Any],
attrs: Mapping[str, Any],
) -> pd.DataFrame:
row_names = attrs["row.names"]
obj = {key: _dataframe_column_transform(val) for key, val in obj.items()}
# Default row names are stored as [R_INT_NA, -len]
default_row_names_len = 2
index: pd.RangeIndex | tuple[str, ...] = (
pd.RangeIndex(1, abs(row_names[1]) + 1)
if (
len(row_names) == default_row_names_len
and isinstance(row_names, np.ma.MaskedArray)
and row_names.mask[0]
)
else tuple(row_names)
)
return pd.DataFrame(obj, columns=obj, index=index)
def _factor_constructor_internal(
obj: np.ndarray[Any, np.dtype[np.integer[Any]]],
attrs: Mapping[str, Any],
*,
ordered: bool,
) -> pd.Categorical:
values = [attrs["levels"][i - 1] if i >= 0 else None for i in obj]
return pd.Categorical(values, attrs["levels"], ordered=ordered)
def factor_constructor(
obj: np.ndarray[Any, np.dtype[np.integer[Any]]],
attrs: Mapping[str, Any],
) -> pd.Categorical:
"""Construct a factor objects."""
return _factor_constructor_internal(obj, attrs, ordered=False)
def ordered_constructor(
obj: np.ndarray[Any, np.dtype[np.integer[Any]]],
attrs: Mapping[str, Any],
) -> pd.Categorical:
"""Contruct an ordered factor."""
return _factor_constructor_internal(obj, attrs, ordered=True)
def ts_constructor(
obj: np.ndarray[Any, Any],
attrs: Mapping[str, Any],
) -> pd.Series[Any]:
"""Construct a time series object."""
start, end, frequency = attrs["tsp"]
frequency = int(frequency)
real_start = Fraction(int(round(start * frequency)), frequency)
real_end = Fraction(int(round(end * frequency)), frequency)
index: np.ndarray[Any, Any] = np.arange(
real_start,
real_end + Fraction(1, frequency),
Fraction(1, frequency),
)
if frequency == 1:
index = index.astype(int)
return pd.Series(obj, index=index)
[docs]
@dataclass
class SrcRef:
"""Reference to a source file location."""
first_line: int
first_byte: int
last_line: int
last_byte: int
first_column: int
last_column: int
first_parsed: int
last_parsed: int
srcfile: SrcFile
def srcref_constructor(
obj: tuple[int, int, int, int, int, int, int, int],
attrs: Mapping[str, Any],
) -> SrcRef:
return SrcRef(*obj, srcfile=attrs["srcfile"])
[docs]
@dataclass
class SrcFile:
"""Source file."""
filename: str
file_encoding: str | None
string_encoding: str | None
def srcfile_constructor(
obj: REnvironment,
attrs: Mapping[str, Any], # noqa: ARG001
) -> SrcFile:
frame = obj.frame
assert frame is not None
filename = frame["filename"][0]
file_encoding = frame.get("encoding")
string_encoding = frame.get("Enc")
return SrcFile(
filename=filename,
file_encoding=file_encoding,
string_encoding=string_encoding,
)
[docs]
@dataclass
class SrcFileCopy(SrcFile):
"""Source file with a copy of its lines."""
lines: Sequence[str]
def srcfilecopy_constructor(
obj: REnvironment,
attrs: Mapping[str, Any], # noqa: ARG001
) -> SrcFileCopy:
frame = obj.frame
assert frame is not None
filename = frame["filename"][0]
file_encoding = frame.get("encoding", (None,))[0]
string_encoding = frame.get("Enc", (None,))[0]
lines = frame["lines"]
return SrcFileCopy(
filename=filename,
file_encoding=file_encoding,
string_encoding=string_encoding,
lines=lines,
)
Constructor = Callable[[Any, Mapping[str, Any]], Any]
ConstructorDict = Mapping[
Union[str, bytes],
Constructor,
]
default_class_map_dict: Final[ConstructorDict] = {
"data.frame": dataframe_constructor,
"factor": factor_constructor,
"ordered": ordered_constructor,
"ts": ts_constructor,
"srcref": srcref_constructor,
"srcfile": srcfile_constructor,
"srcfilecopy": srcfilecopy_constructor,
}
#: Default mapping of constructor functions.
DEFAULT_CLASS_MAP: Final = MappingProxyType(default_class_map_dict)
[docs]
class Converter(abc.ABC):
"""Interface of a class converting R objects in Python objects."""
[docs]
@abc.abstractmethod
def convert(self, data: parser.RData | parser.RObject) -> Any: # noqa: ANN401
"""Convert a R object to a Python one."""
@dataclass
class UnresolvedReference:
references: MutableMapping[int, Any]
index: int
[docs]
class SimpleConverter(Converter):
"""
Class converting R objects to Python objects.
Args:
constructor_dict:
Dictionary mapping names of R classes to constructor functions with
the following prototype:
.. code-block :: python
def constructor(obj, attrs):
...
This dictionary can be used to support custom R classes. By
default, the dictionary used is
:data:`~rdata.conversion._conversion.DEFAULT_CLASS_MAP`
which has support for several common classes.
default_encoding:
Default encoding used for strings with unknown encoding. If `None`,
the one stored in the file will be used, or ASCII as a fallback.
force_default_encoding:
Use the default encoding even if the strings specify other
encoding.
global_environment: Global environment to use. By default is an empty
environment.
base_environment: Base environment to use. By default is an empty
environment.
"""
def __init__(
self,
constructor_dict: ConstructorDict = DEFAULT_CLASS_MAP,
*,
default_encoding: str | None = None,
force_default_encoding: bool = False,
global_environment: MutableMapping[str, Any] | None = None,
base_environment: MutableMapping[str, Any] | None = None,
) -> None:
self.constructor_dict = constructor_dict
self.default_encoding = default_encoding
self.force_default_encoding = force_default_encoding
self.global_environment = REnvironment(
{} if global_environment is None
else global_environment,
)
self.base_environment = REnvironment(
{} if base_environment is None
else base_environment,
)
self.empty_environment: Mapping[str, Any] = REnvironment({})
self._reset()
def _reset(self) -> None:
self.references: MutableMapping[int, Any] = {}
self.default_encoding_used = self.default_encoding
[docs]
@override
def convert(
self,
data: parser.RData | parser.RObject,
) -> Any:
self._reset()
return self._convert_next(data)
def _convert_next( # noqa: C901, PLR0912, PLR0915
self,
data: parser.RData | parser.RObject,
) -> Any: # noqa: ANN401
"""Convert a R object to a Python one."""
obj: parser.RObject
if isinstance(data, parser.RData):
obj = data.object
if self.default_encoding is None:
self.default_encoding_used = data.extra.encoding
else:
obj = data
attrs = convert_attrs(obj, self._convert_next)
reference_id = id(obj)
# Return the value if previously referenced
value: Any = self.references.get(id(obj))
if value is not None:
pass
if obj.info.type == parser.RObjectType.SYM:
# Return the internal string
value = convert_symbol(obj, self._convert_next)
elif obj.info.type == parser.RObjectType.LIST:
# Expand the list and process the elements
value = convert_list(obj, self._convert_next)
elif obj.info.type == parser.RObjectType.CLO:
assert obj.tag is not None
assert obj.attributes is not None
environment = self._convert_next(obj.tag)
formals = self._convert_next(obj.value[0])
body = self._convert_next(obj.value[1])
attributes = self._convert_next(obj.attributes)
value = RFunction(
environment=environment,
formals=formals,
body=body,
attributes=attributes,
)
elif obj.info.type == parser.RObjectType.ENV:
# Return a ChainMap of the environments
value = convert_env(obj, self._convert_next)
elif obj.info.type == parser.RObjectType.LANG:
# Expand the list and process the elements, returning a
# special object
rlanguage_list = convert_list(obj, self._convert_next)
assert isinstance(rlanguage_list, list)
attributes = self._convert_next(
obj.attributes,
) if obj.attributes else {}
value = RLanguage(rlanguage_list, attributes)
elif obj.info.type in {
parser.RObjectType.SPECIAL,
parser.RObjectType.BUILTIN,
}:
value = RBuiltin(name=obj.value.decode("ascii"))
elif obj.info.type == parser.RObjectType.CHAR:
# Return the internal string
value = convert_char(
obj,
default_encoding=self.default_encoding_used,
force_default_encoding=self.force_default_encoding,
)
elif obj.info.type in {
parser.RObjectType.LGL,
parser.RObjectType.INT,
parser.RObjectType.REAL,
parser.RObjectType.CPLX,
}:
# Return the internal array
value = convert_array(obj, attrs=attrs)
elif obj.info.type == parser.RObjectType.STR:
# Convert the internal strings
value = np.array([self._convert_next(o) for o in obj.value])
elif obj.info.type == parser.RObjectType.VEC:
# Convert the internal objects
value = convert_vector(obj, self._convert_next, attrs=attrs)
elif obj.info.type == parser.RObjectType.EXPR:
rexpression_list = convert_vector(
obj,
self._convert_next,
attrs=attrs,
)
assert isinstance(rexpression_list, list)
# Convert the internal objects returning a special object
value = RExpression(rexpression_list)
elif obj.info.type == parser.RObjectType.BCODE:
value = RBytecode(
code=self._convert_next(obj.value[0]),
constants=[self._convert_next(c) for c in obj.value[1]],
attributes=attrs,
)
elif obj.info.type == parser.RObjectType.EXTPTR:
value = RExternalPointer(
protected=self._convert_next(obj.value[0]),
tag=self._convert_next(obj.value[1]),
)
elif obj.info.type == parser.RObjectType.S4:
value = SimpleNamespace(**attrs)
elif obj.info.type == parser.RObjectType.BASEENV:
value = self.base_environment
elif obj.info.type == parser.RObjectType.EMPTYENV:
value = self.empty_environment
elif obj.info.type == parser.RObjectType.MISSINGARG:
value = NotImplemented
elif obj.info.type == parser.RObjectType.GLOBALENV:
value = self.global_environment
elif obj.info.type == parser.RObjectType.REF:
# Return the referenced value
value = self.references.get(id(obj.referenced_object))
if value is None:
reference_id = id(obj.referenced_object)
assert obj.referenced_object is not None
self.references[reference_id] = UnresolvedReference(
self.references,
reference_id,
)
value = self._convert_next(obj.referenced_object)
elif obj.info.type == parser.RObjectType.NILVALUE:
value = None
else:
msg = f"Type {obj.info.type} not implemented"
raise NotImplementedError(msg)
if obj.info.object and attrs is not None:
classname = attrs.get("class", ())
for i, c in enumerate(classname):
constructor = self.constructor_dict.get(c, None)
new_value = (
constructor(value, attrs)
if constructor
else NotImplemented
)
if new_value is NotImplemented:
missing_msg = (
f"Missing constructor for R class \"{c}\". "
)
if len(classname) > (i + 1):
solution_msg = (
f"The constructor for class "
f"\"{classname[i+1]}\" will be "
f"used instead."
)
else:
solution_msg = (
"The underlying R object is "
"returned instead."
)
warnings.warn(
missing_msg + solution_msg,
stacklevel=1,
)
else:
value = new_value
break
self.references[reference_id] = value
return value
[docs]
def convert(
data: parser.RData | parser.RObject,
constructor_dict: ConstructorDict = DEFAULT_CLASS_MAP,
*,
default_encoding: str | None = None,
force_default_encoding: bool = False,
global_environment: MutableMapping[str, Any] | None = None,
base_environment: MutableMapping[str, Any] | None = None,
) -> Any: # noqa: ANN401
"""
Use the default converter (:func:`SimpleConverter`) to convert the data.
Args:
data: Parsed data.
constructor_dict: Dictionary mapping names of R classes to constructor
functions with the following prototype:
.. code-block :: python
def constructor(obj, attrs):
...
This dictionary can be used to support custom R classes. By
default, the dictionary used is
:data:`~rdata.conversion._conversion.DEFAULT_CLASS_MAP`
which has support for several common classes.
default_encoding: Default encoding used for strings with unknown
encoding. If `None`, the one stored in the file will be used, or
ASCII as a fallback.
force_default_encoding:
Use the default encoding even if the strings specify other
encoding.
global_environment: Global environment to use. By default is an empty
environment.
base_environment: Base environment to use. By default is an empty
environment.
Examples:
Parse one of the included examples, containing a vector
>>> import rdata
>>>
>>> parsed = rdata.parser.parse_file(
... rdata.TESTDATA_PATH / "test_vector.rda")
>>> converted = rdata.conversion.convert(parsed)
>>> converted
{'test_vector': array([1., 2., 3.])}
Parse another example, containing a dataframe
>>> import rdata
>>>
>>> parsed = rdata.parser.parse_file(
... rdata.TESTDATA_PATH / "test_dataframe.rda")
>>> converted = rdata.conversion.convert(parsed)
>>> converted
{'test_dataframe': class value
1 a 1
2 b 2
3 b 3}
"""
return SimpleConverter(
constructor_dict=constructor_dict,
default_encoding=default_encoding,
force_default_encoding=force_default_encoding,
global_environment=global_environment,
base_environment=base_environment,
).convert(data)
