1
2
3
4
5
6
7
8
9
10
11
12
13
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
60
61
62
|
"""Utilities for testing with R files."""
from __future__ import annotations
import subprocess
import tempfile
from typing import Any, Protocol
R_CODE_PREFIX = """::: """
class HasDoc(Protocol):
"""Python object having a docstring."""
__doc__: str | None
def get_data_source(
function_or_class: HasDoc,
*,
prefix: str = R_CODE_PREFIX,
) -> str:
"""Get the part of the docstring containing the data source."""
doc = function_or_class.__doc__
if doc is None:
return ""
source = ""
for line in doc.splitlines(keepends=True):
stripped_line = line.lstrip()
if stripped_line.startswith(prefix):
source += stripped_line.removeprefix(prefix)
return source
def execute_r_data_source(
function_or_class: HasDoc,
*,
prefix: str = R_CODE_PREFIX,
**kwargs: Any, # noqa: ANN401
) -> None:
"""Execute R data source."""
source = get_data_source(
function_or_class,
prefix=prefix,
)
if not source:
return
inits = ""
for key, value in kwargs.items():
inits += f"{key} <- {value!r}\n"
source = inits + source
with tempfile.NamedTemporaryFile("w") as file:
file.write(source)
file.flush()
subprocess.check_call(
["Rscript", file.name], # noqa: S603, S607
)
|
