rpy2 中 Python 对象到 R 矩阵的高效与安全转换指南

import rpy2.robjects as robjects
from rpy2.robjects import numpy2ri
import numpy as np

# 全局激活转换
numpy2ri.activate()

data = np.array([1, 2, 3, 4])
# 当 numpy2ri 激活时，robjects.r.matrix 会自动将 NumPy 数组转换为 R 向量
r_matrix = robjects.r.matrix(data, nrow=2, ncol=2)
print(type(r_matrix)) # 输出: <class 'rpy2.robjects.vectors.IntMatrix'>

# 全局停用转换
numpy2ri.deactivate()

import rpy2.robjects as robjects
from rpy2.robjects import numpy2ri, pandas2ri
from rpy2.robjects.conversion import localconverter
import numpy as np
import pandas as pd

# 示例数据
python_np_array = np.array([[1, 2], [3, 4]], dtype=int)
python_pd_dataframe = pd.DataFrame({'A': [1, 2], 'B': [3, 4]})

print(f"原始 NumPy 数组类型: {type(python_np_array)}")

# 使用 localconverter 进行局部转换
with localconverter(robjects.default_converter + numpy2ri.converter):
    # 在此上下文块内，NumPy 数组会自动转换为 R 向量或矩阵
    # robjects.r.matrix 会将 python_np_array 视为一个 R 向量进行处理
    r_matrix_from_np = robjects.r.matrix(python_np_array, nrow=2, ncol=2)
    print(f"转换后的 R 矩阵类型 (通过 numpy2ri): {type(r_matrix_from_np)}")

# 也可以将 numpy2ri.converter 作为单独的上下文
with localconverter(robjects.default_converter + numpy2ri.converter):
    # 直接将 NumPy 数组转换为 R 矩阵，如果转换器支持
    # 注意：这里如果直接传递 np 数组给一个期望 R 对象的函数，
    # 转换器会尝试将其转换为最合适的 R 对象。
    # 对于 r.matrix，它仍然期望一个扁平化的 R 向量。
    # 更直接的 NumPy 矩阵到 R 矩阵的转换通常发生在 rpy2 内部函数调用时。
    # 但对于 robjects.r.matrix，它仍按 R 的语义工作。
    pass # 示例仅为演示上下文

# 确保 Python 对象类型兼容
请注意，`robjects.r.matrix()` 函数在 R 中期望接收一个扁平化的向量作为其第一个参数。当 `numpy2ri.converter` 激活时，它会将 NumPy 数组自动转换为 R 向量，然后 `robjects.r.matrix()` 再用这个向量来构建矩阵。

如果您有一个 NumPy 矩阵，并希望将其转换为 R 矩阵，最关键的是确保 `numpy2ri.converter` 处于活动状态，并且您以 R 函数期望的方式提供数据。

import rpy2.robjects as robjects
from rpy2.robjects import numpy2ri, pandas2ri
from rpy2.robjects.conversion import localconverter, Converter
from rpy2.robjects.packages import importr
import numpy as np
import networkx as nx

# 导入必要的R包
base = importr('base')
graph = importr('graph') # 假设 graphNEL 需要这个包

# 创建一个默认的转换器实例，避免在每次循环中重复创建
# 确保在整个函数执行过程中，numpy2ri.converter 可以在需要时被激活
my_converter = robjects.default_converter + numpy2ri.converter + pandas2ri.converter

def sample_graphs(mpgraph, n_graphs=10, equal_weights=False):
    graphs = []
    if nx.is_directed_acyclic_graph(nx.DiGraph(mpgraph)):
        graphs.append((mpgraph.copy(), n_graphs))
    else:
        n_vars = mpgraph.shape[0]

        # 确保 addBgKnowledge 可用，这里假设它是一个 R 函数
        addBgKnowledge = robjects.r['addBgKnowledge']
        r_as = robjects.r['as'] # R 中的 as 函数，用于类型转换

        for _ in range(n_graphs):
            graph = mpgraph.copy() # 假设 graph 是一个 NumPy 数组
            undirected_u, undirected_v = np.nonzero(np.triu(graph == graph.T) & (graph == 1))

            while len(undirected_u) > 0:
                selected_edge_idx = np.random.randint(0, len(undirected_u))
                u, v = undirected_u[selected_edge_idx], undirected_v[selected_edge_idx]
                if np.random.rand() < 0.5:
                    u, v = v, u

                # 使用 localconverter 确保 numpy2ri 在此代码块中激活
                with localconverter(my_converter):
                    # 当 numpy2ri 激活时，Python 的 'graph' (NumPy 数组) 会被转换为 R 向量
                    # 然后 robjects.r.matrix 使用这个 R 向量构建 R 矩阵
                    cpgraph_r_matrix = robjects.r.matrix(graph, nrow=n_vars, ncol=n_vars)
                    print(f"cpgraph 类型: {type(cpgraph_r_matrix)}") # 应该输出 <class 'rpy2.robjects.vectors.IntMatrix'>

                    cpgraph_r_matrix.rownames = robjects.StrVector([str(i) for i in range(n_vars)])
                    cpgraph_r_matrix.colnames = robjects.StrVector([str(i) for i in range(n_vars)])

                    # 将 R 矩阵转换为 graphNEL 对象
                    # 注意：'graphNEL' 类型转换需要 R 的 'graph' 包
                    # 并且需要确保 cpgraph_r_matrix 是一个合适的 R 矩阵对象
                    cpgraph_graphNEL = r_as(cpgraph_r_matrix, 'graphNEL')

                    # 调用 R 函数 addBgKnowledge，并再次使用 r_as 转换回 Python 矩阵
                    # 这里假设 addBgKnowledge 返回一个 R 矩阵，并且 numpy2ri 再次帮助转换
                    graph = np.asarray(r_as(addBgKnowledge(cpgraph_graphNEL, x=[str(u)], y=[str(v)]), 'matrix'))
                    graph = graph.astype(int) # 确保转换回 NumPy int 矩阵

                undirected_u, undirected_v = np.nonzero(np.triu(graph == graph.T) & (graph == 1))

            found = False
            for idx, (comp_graph, weight) in enumerate(graphs):
                if (comp_graph == graph).all():
                    graphs[idx] = (graph, weight + 1)
                    found = True
                    break

            if not found:
                graphs.append((graph, 1))

    if equal_weights:
        graphs = [(graph, 1 / len(graphs)) for graph, _ in graphs]
    else:
        graphs = [(graph, w / n_graphs) for graph, w in graphs]
    return graphs

# 示例调用 (需要根据实际环境模拟 mpgraph 和 addBgKnowledge)
# mpgraph = np.random.randint(0, 2, size=(5, 5))
# # 确保 mpgraph 至少是无向的，并且有循环以进入 else 分支
# mpgraph = (mpgraph + mpgraph.T) // 2
# np.fill_diagonal(mpgraph, 0)
#
# # 模拟 R 的 addBgKnowledge 函数
# robjects.r('''
#     library(graph)
#     addBgKnowledge <- function(g, x, y) {
#         # 这是一个简化的模拟，实际行为会更复杂
#         # 假设它返回一个修改后的矩阵
#         adj_matrix <- as(g, "matrix")
#         # 模拟添加边，例如 x -> y
#         # 这里需要根据 graphNEL 的结构来修改，只是一个示意
#         # 实际操作会涉及到 graph 包的函数，例如 addEdge
#         # 为了简单，我们直接修改矩阵并返回
#         if (length(x) > 0 && length(y) > 0) {
#             x_idx <- as.integer(x[1]) + 1 # R 是1基索引
#             y_idx <- as.integer(y[1]) + 1
#             if (x_idx <= nrow(adj_matrix) && y_idx <= ncol(adj_matrix)) {
#                 adj_matrix[x_idx, y_idx] <- 1
#             }
#         }
#         return(adj_matrix)
#     }
# ''')
#
# # 确保 addBgKnowledge 在 R 环境中可用
# # 重新获取 R 函数对象
# addBgKnowledge = robjects.r['addBgKnowledge']
#
# # 调用示例函数
# # result_graphs = sample_graphs(mpgraph, n_graphs=2)
# # print(result_graphs)