跳转至

Utils.reader(读取函数) 模块

ppsci.utils.reader

load_csv_file(file_path, keys, alias_dict=None, delimiter=',', encoding='utf-8')

Load *.csv file and fetch data as given keys.

Parameters:

Name Type Description Default
file_path str

CSV file path.

required
keys Tuple[str, ...]

Required fetching keys.

required
alias_dict Optional[Dict[str, str]]

Alias for keys, i.e. {inner_key: outer_key}. Defaults to None.

None
encoding str

Encoding code when open file. Defaults to "utf-8".

'utf-8'

Returns:

Type Description
Dict[str, ndarray]

Dict[str, np.ndarray]: Loaded data in dict.

Source code in ppsci/utils/reader.py
def load_csv_file(
    file_path: str,
    keys: Tuple[str, ...],
    alias_dict: Optional[Dict[str, str]] = None,
    delimiter: str = ",",
    encoding: str = "utf-8",
) -> Dict[str, np.ndarray]:
    """Load *.csv file and fetch data as given keys.

    Args:
        file_path (str): CSV file path.
        keys (Tuple[str, ...]): Required fetching keys.
        alias_dict (Optional[Dict[str, str]]): Alias for keys,
            i.e. {inner_key: outer_key}. Defaults to None.
        encoding (str, optional): Encoding code when open file. Defaults to "utf-8".

    Returns:
        Dict[str, np.ndarray]: Loaded data in dict.
    """
    if alias_dict is None:
        alias_dict = {}

    try:
        # read all data from csv file
        with open(file_path, "r", encoding=encoding) as csv_file:
            reader = csv.DictReader(csv_file, delimiter=delimiter)
            raw_data = collections.defaultdict(list)
            for _, line_dict in enumerate(reader):
                for key, value in line_dict.items():
                    raw_data[key].append(value)
    except FileNotFoundError as e:
        raise e

    # convert to numpy array
    data_dict = {}
    for key in keys:
        fetch_key = alias_dict[key] if key in alias_dict else key
        if fetch_key not in raw_data:
            raise KeyError(f"fetch_key({fetch_key}) do not exist in raw_data.")
        data_dict[key] = np.asarray(raw_data[fetch_key])
        if not np.issubdtype(data_dict[key].dtype, np.integer):
            data_dict[key] = data_dict[key].astype(paddle.get_default_dtype())
        data_dict[key] = data_dict[key].reshape([-1, 1])

    return data_dict

load_mat_file(file_path, keys, alias_dict=None)

Load *.mat file and fetch data as given keys.

Parameters:

Name Type Description Default
file_path str

Mat file path.

required
keys Tuple[str, ...]

Required fetching keys.

required
alias_dict Optional[Dict[str, str]]

Alias for keys, i.e. {original_key: original_key}. Defaults to None.

None

Returns:

Type Description
Dict[str, ndarray]

Dict[str, np.ndarray]: Loaded data in dict.

Source code in ppsci/utils/reader.py
def load_mat_file(
    file_path: str, keys: Tuple[str, ...], alias_dict: Optional[Dict[str, str]] = None
) -> Dict[str, np.ndarray]:
    """Load *.mat file and fetch data as given keys.

    Args:
        file_path (str): Mat file path.
        keys (Tuple[str, ...]): Required fetching keys.
        alias_dict (Optional[Dict[str, str]]): Alias for keys,
            i.e. {original_key: original_key}. Defaults to None.

    Returns:
        Dict[str, np.ndarray]: Loaded data in dict.
    """

    if alias_dict is None:
        alias_dict = {}

    try:
        # read all data from mat file
        raw_data = sio.loadmat(file_path)
    except FileNotFoundError as e:
        raise e

    # convert to numpy array
    data_dict = {}
    for key in keys:
        fetch_key = alias_dict[key] if key in alias_dict else key
        if fetch_key not in raw_data:
            raise KeyError(f"fetch_key({fetch_key}) do not exist in raw_data.")
        data_dict[key] = np.asarray(raw_data[fetch_key])
        if not np.issubdtype(data_dict[key].dtype, np.integer):
            data_dict[key] = data_dict[key].astype(paddle.get_default_dtype())
        data_dict[key] = data_dict[key].reshape([-1, 1])

    return data_dict

load_npz_file(file_path, keys, alias_dict=None)

Load *.npz file and fetch data as given keys.

Parameters:

Name Type Description Default
file_path str

Npz file path.

required
keys Tuple[str, ...]

Required fetching keys.

required
alias_dict Optional[Dict[str, str]]

Alias for keys, i.e. {original_key: original_key}. Defaults to None.

None

Returns:

Type Description
Dict[str, ndarray]

Dict[str, np.ndarray]: Loaded data in dict.

Source code in ppsci/utils/reader.py
def load_npz_file(
    file_path: str, keys: Tuple[str, ...], alias_dict: Optional[Dict[str, str]] = None
) -> Dict[str, np.ndarray]:
    """Load *.npz file and fetch data as given keys.

    Args:
        file_path (str): Npz file path.
        keys (Tuple[str, ...]): Required fetching keys.
        alias_dict (Optional[Dict[str, str]]): Alias for keys,
            i.e. {original_key: original_key}. Defaults to None.

    Returns:
        Dict[str, np.ndarray]: Loaded data in dict.
    """

    if alias_dict is None:
        alias_dict = {}

    try:
        # read all data from npz file
        raw_data = np.load(file_path, allow_pickle=True)
    except FileNotFoundError as e:
        raise e

    # convert to numpy array
    data_dict = {}
    for key in keys:
        fetch_key = alias_dict[key] if key in alias_dict else key
        if fetch_key not in raw_data:
            raise KeyError(f"fetch_key({fetch_key}) do not exist in raw_data.")
        data_dict[key] = np.asarray(raw_data[fetch_key])
        if data_dict[key].dtype in (np.float16, np.float32, np.float64):
            data_dict[key] = data_dict[key].astype(paddle.get_default_dtype())

    return data_dict

load_vtk_file(filename_without_timeid, time_step, time_index, input_keys, label_keys)

Load coordinates and attached label from the *.vtu file.

Parameters:

Name Type Description Default
filename_without_timeid str

File name without time id.

required
time_step float

Physical time step.

required
time_index Tuple[int, ...]

Physical time indexes.

required
input_keys Tuple[str, ...]

Input coordinates name keys.

required
label_keys Optional[Tuple[str, ...]]

Input label name keys.

required

Returns:

Type Description
Dict[str, ndarray]

Dict[str, np.ndarray]: Input coordinates dict, label coordinates dict

Source code in ppsci/utils/reader.py
def load_vtk_file(
    filename_without_timeid: str,
    time_step: float,
    time_index: Tuple[int, ...],
    input_keys: Tuple[str, ...],
    label_keys: Optional[Tuple[str, ...]],
) -> Dict[str, np.ndarray]:
    """Load coordinates and attached label from the *.vtu file.

    Args:
        filename_without_timeid (str): File name without time id.
        time_step (float): Physical time step.
        time_index (Tuple[int, ...]): Physical time indexes.
        input_keys (Tuple[str, ...]): Input coordinates name keys.
        label_keys (Optional[Tuple[str, ...]]): Input label name keys.

    Returns:
        Dict[str, np.ndarray]: Input coordinates dict, label coordinates dict
    """
    input_dict = {var: [] for var in input_keys}
    label_dict = {var: [] for var in label_keys}
    for index in time_index:
        file = filename_without_timeid + f"{index}.vtu"
        mesh = meshio.read(file)
        n = mesh.points.shape[0]
        i = 0
        for key in input_dict:
            if key == "t":
                input_dict[key].append(
                    np.full((n, 1), index * time_step, paddle.get_default_dtype())
                )
            else:
                input_dict[key].append(
                    mesh.points[:, i].reshape(n, 1).astype(paddle.get_default_dtype())
                )
                i += 1
        for i, key in enumerate(label_dict):
            label_dict[key].append(
                np.array(mesh.point_data[key], paddle.get_default_dtype())
            )
    for key in input_dict:
        input_dict[key] = np.concatenate(input_dict[key])
    for key in label_dict:
        label_dict[key] = np.concatenate(label_dict[key])

    return input_dict, label_dict

load_vtk_with_time_file(file)

Temporary interface for points cloud, will be banished sooner.

Parameters:

Name Type Description Default
file str

Input file name.

required

Returns:

Type Description
Dict[str, ndarray]

Dict[str, np.ndarray]: Input coordinates dict.

Source code in ppsci/utils/reader.py
def load_vtk_with_time_file(file: str) -> Dict[str, np.ndarray]:
    """Temporary interface for points cloud, will be banished sooner.

    Args:
        file (str): Input file name.

    Returns:
        Dict[str, np.ndarray]: Input coordinates dict.
    """
    mesh = meshio.read(file)
    n = mesh.points.shape[0]
    t = np.array(mesh.point_data["time"])
    x = mesh.points[:, 0].reshape(n, 1)
    y = mesh.points[:, 1].reshape(n, 1)
    z = mesh.points[:, 2].reshape(n, 1)
    input_dict = {"t": t, "x": x, "y": y, "z": z}
    return input_dict