Spaces:
Sleeping
Sleeping
File size: 3,226 Bytes
f54e1d4 | 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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | from typing import Any, Optional, Union
class DataFact:
def __init__(self):
# 用 dict 描述我们的 fact, 包含的 keys
self.type: str = ""
self.subtype: str = ""
self.data_points: dict = {}
self.score: float = 0.0
self.annotation: str = ""
self.reason: str = ""
self.types = []
def set_value(self,
subtype: Optional[str] = None,
data_points: Optional[dict] = None,
score: Optional[float] = None,
annotation: Optional[str] = None,
reason: Optional[str] = None
):
""" 设置各变量值 """
if subtype is not None:
if subtype in self.types:
self.subtype = subtype
else:
print(f"Invalid type: {subtype}.")
if data_points is not None:
self.data_points = data_points
if score is not None:
self.score = score
if annotation is not None:
self.annotation = annotation
if reason is not None:
self.reason = reason
def get_json(self):
""" 返回 json 格式 """
formated_json = {
"type": self.type,
"subtype": self.subtype,
"data_points": self.data_points,
"score": round(self.score, 2),
"annotation": self.annotation,
"reason": self.reason
}
return formated_json
class DataFactGenerator:
def __init__(self, data: dict):
self.data = data
self.data_columns: dict[str, Any] = self.data["data"]["columns"]
self.tabular_data: list[dict[str, Any]] = self.data["data"]["data"] # 原始数据
self.grouped_data = divide_data_by_group(self.data_columns, self.tabular_data)
# metadata
self.x_column = self.data_columns[0]["name"]
self.y_column = self.data_columns[1]["name"]
self.group_column = self.data_columns[2]["name"] if len(self.data_columns) > 2 and self.data_columns[2]["data_type"] in ["categorical", "temporal"] else None
is_temporal = False
col = self.data_columns[0]
if col["data_type"] == "temporal":
is_temporal = True
else:
is_temporal = False
self.is_temporal = is_temporal
def divide_data_by_group(data_columns: list[dict[str, Any]], data: list[dict[str, Any]]) -> dict[str, dict[str, list[Any]]]:
x_column = data_columns[0]["name"]
y_column = data_columns[1]["name"]
group_column = data_columns[2]["name"] if len(data_columns) > 2 and data_columns[2]["data_type"] in ["categorical", "temporal"] else None
grouped_data = {}
for idx, row in enumerate(data):
group_value = row.get(group_column, "")
if group_value not in grouped_data.keys():
grouped_data[group_value] = {
"indices": [],
"x_list": [],
"y_list": []
}
grouped_data[group_value]["indices"].append(idx)
grouped_data[group_value]["x_list"].append(row[x_column])
grouped_data[group_value]["y_list"].append(row[y_column])
return grouped_data
|