Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 82 additions & 1 deletion src/ansys/fluent/core/utils/dump_session_data.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
"""Module providing dump session data functionality."""
from pathlib import Path
import pickle
from typing import Optional
from typing import Optional, Union

import numpy as np

from ansys.fluent.core.services.field_data import SurfaceDataType


def dump_session_data(
Expand Down Expand Up @@ -84,7 +89,83 @@ def dump_session_data(
transaction.add_vector_fields_request(
surface_ids=surfaces_id, field_name="velocity"
)
transaction.add_pathlines_fields_request(
surface_ids=surfaces_id, field_name="velocity-magnitude"
)
session_data["fields"] = transaction.get_fields()

with open(file_path, "wb") as pickle_obj:
pickle.dump(session_data, pickle_obj)


class DumpDataReader:
def __init__(self, file_path: str):
with open(
str(Path(file_path).resolve()),
"rb",
) as pickle_obj:
self._session_data = pickle.load(pickle_obj)

def get_session_data(self):
return self._session_data

def get_surface_data(self, surface_ids, data_type) -> list[Union[np.array, None]]:
tag_id = (("type", "surface-data"),)

enum_to_field_name = {
SurfaceDataType.FacesConnectivity: "faces",
SurfaceDataType.Vertices: "vertices",
SurfaceDataType.FacesCentroid: "centroid",
SurfaceDataType.FacesNormal: "face-normal",
}

surface_data = []
for surface_id in surface_ids:
surface_data.append(
self._session_data["fields"][tag_id][surface_id][
enum_to_field_name[data_type]
]
)

return surface_data

def get_scalar_field_data(
self, surface_ids, data_location, provide_boundary_values, field_name
) -> list[Union[np.array, None]]:
tag_id = (
("type", "scalar-field"),
("dataLocation", data_location),
("boundaryValues", provide_boundary_values),
)

scalar_field_data = []
for surface_id in surface_ids:
scalar_field_data.append(
self._session_data["fields"][tag_id][surface_id][field_name]
)

return scalar_field_data

def get_vector_field_data(
self, surface_ids, field_name
) -> list[Union[np.array, None]]:
tag_id = (("type", "vector-field"),)

vector_field_data = []
for surface_id in surface_ids:
vector_field_data.append(
self._session_data["fields"][tag_id][surface_id][field_name]
)

return vector_field_data

def get_pathlines_data(
self, surface_ids, field_name
) -> list[Union[np.array, None]]:
tag_id = (("type", "pathlines-field"), ("field", field_name))

pathlines_data = []
for surface_id in surface_ids:
pathlines_data.append(self._session_data["fields"][tag_id][surface_id])

return pathlines_data