Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Extended magnetic mapping #222

Closed
wants to merge 8 commits into from
Closed

Extended magnetic mapping #222

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
f0e5394
Extended magnetic mapping
AreWeDreaming Nov 30, 2022
2f0e0b3
Fixed some horrible math errors from last commit
AreWeDreaming Jan 5, 2023
a634444
Fixed error in rho_tor to rho_pol conversion
AreWeDreaming Feb 16, 2023
1d7ab5e
Merged master into this branch
AreWeDreaming Feb 16, 2023
a35e79d
Fixed some bugs, but the result is still wrong
AreWeDreaming Feb 16, 2023
df2cf6b
Fixed broken psi <-> conversion
AreWeDreaming Feb 16, 2023
989b295
Merge remote-tracking branch 'origin/master' into mapping_to_from_rho…
AreWeDreaming Jul 13, 2023
8f1568a
Merge remote-tracking branch 'origin/master' into mapping_to_from_rho…
AreWeDreaming Oct 21, 2023
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
142 changes: 125 additions & 17 deletions omas/omas_physics.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,8 +213,72 @@ def equilibrium_ggd_to_rectangular(ods, time_index=None, resolution=None, method
profiles_2d['grid.dim2'] = z
return ods_n

@add_to__ODS__
@preprocess_ods('equilibrium')
def add_rho_pol_norm_to_equilbrium_profiles_1d_ods(ods, time_index):
try:
if (len(ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["rho_pol-norm"]) ==
len(ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["psi"])):
return
else:
raise LookupError
except LookupError:
ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["rho_pol_norm"] = (
ods.map_pol_flux_to_flux_coordinate(ods, time_index, "rho_pol_norm",
ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["psi"])
)

def mask_SOL(ods, time_index, psi_values):
"""
Returns a numpy array of `dtype=bool`/
The array is true for all values inside and on the LCFS
:param ods: input ods

:param time_index: time slices to process

:param values: Psi values that need to masked

:return: mask that is True for values inside and on the LCFS
"""
if (ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_axis"] <
ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_boundary"]):
return psi_values <= ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_boundary"]
else:
return psi_values >= ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_boundary"]

@add_to__ODS__
@preprocess_ods('equilibrium')
def add_phi_to_equilbrium_profiles_1d_ods(ods, time_index):
"""
Adds `profiles_1d.phi` to an ODS using q
:param ods: input ods

:param time_index: time slices to process
"""
try:
if (len(ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["phi"]) ==
len(ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["psi"])):
return
else:
raise LookupError
except (LookupError, ValueError):
from scipy.interpolate import InterpolatedUnivariateSpline
#TODO:
# - Any cocos needed here?
psi = ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["psi"]
mask = mask_SOL(ods, time_index, psi)
q_spline = InterpolatedUnivariateSpline(
ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["psi"][mask],
ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["q"][mask])
phi_spline = q_spline.antiderivative(1)
ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["phi"] = numpy.zeros(psi.shape)
ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["phi"][mask] = (
phi_spline(ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["psi"][mask]))
ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["phi"][mask==False] = numpy.inf


def map_flux_coordinate_to_pol_flux(ods, time_index, origin, values):
import numpy as np
"""
Maps from one magnetic coordinate system to psi
:param ods: input ods
Expand All @@ -227,15 +291,31 @@ def map_flux_coordinate_to_pol_flux(ods, time_index, origin, values):

:return: Transformed values
"""
if origin == "rho_pol":
return (
values**2
* (
ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_sep"]
- ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_axis"]
)
+ ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_axis"]
)
if origin == "rho_pol_norm":
return (values**2 * (ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_boundary"]
- ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_axis"])
+ ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_axis"])
elif origin == "rho_tor_norm":
phi = values**2
phi *= map_pol_flux_to_flux_coordinate(ods, time_index, "phi",
np.array([ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_boundary"]]))
return map_flux_coordinate_to_pol_flux(ods, time_index, "phi", phi)
elif origin == "phi":
from scipy.interpolate import InterpolatedUnivariateSpline
psi_grid = ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["psi"]
psi_mask = mask_SOL(ods, time_index, psi_grid)
phi_grid = ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["phi"][psi_mask]
if phi_grid[-1] < phi_grid[0]:
psi_spl = InterpolatedUnivariateSpline(phi_grid[psi_mask][::-1], psi_grid[psi_mask][::-1])
else:
psi_spl = InterpolatedUnivariateSpline(phi_grid[psi_mask], psi_grid[psi_mask])
phi_min = np.min(phi_grid)
phi_max = np.max(phi_grid)
values_mask = np.logical_and(values >= phi_min, values <= phi_max)
psi = np.zeros(values.shape)
psi[:] = np.nan
psi[values_mask] = psi_spl(values[values_mask])
return psi
else:
raise NotImplementedError(f"Conversion from {origin} not yet implemented.")

Expand All @@ -255,14 +335,42 @@ def map_pol_flux_to_flux_coordinate(ods, time_index, destination, values):

:return: Transformed values
"""
if destination == "rho_pol":
return np.sqrt(
(values - ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_axis"])
/ (
ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_boundary"]
- ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_axis"]
)
)
if destination == "rho_pol_norm":
return np.sqrt((values - ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_axis"]) /
(ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_boundary"]
- ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_axis"]))
elif destination == "rho_tor_norm":
mask = mask_SOL(ods, time_index, values)
phi = map_pol_flux_to_flux_coordinate(ods, time_index, "phi", values[mask])
rho_tor_norm = np.zeros(values.shape)
phi_boundary = map_pol_flux_to_flux_coordinate(ods, time_index, "phi",
np.array([ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_boundary"]]))
rho_tor_norm[mask] = np.sqrt(phi / phi_boundary)
rho_tor_norm[mask==False] = np.inf
return rho_tor_norm
elif destination == "phi":
from scipy.interpolate import InterpolatedUnivariateSpline
psi_grid = ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["psi"]
psi_grid_mask = mask_SOL(ods, time_index, psi_grid)
try:
phi_spl = InterpolatedUnivariateSpline(psi_grid[psi_grid_mask],
ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["phi"][psi_grid_mask])
except ValueError:
add_phi_to_equilbrium_profiles_1d_ods(ods, time_index)
phi_spl = InterpolatedUnivariateSpline(psi_grid[psi_grid_mask],
ods["equilibrium"]["time_slice"][time_index]["profiles_1d"]["phi"][psi_grid_mask])
mask = mask_SOL(ods, time_index, values)
phi = np.zeros(values.shape)
phi[mask] = phi_spl(values[mask])
phi_bound = phi_spl(ods["equilibrium"]["time_slice"][time_index]["global_quantities"]["psi_boundary"])
phi[mask==False] = np.inf * np.sign(phi_bound)
wrong_sign_mask = phi_bound * phi < 0
if np.any(wrong_sign_mask):
if np.any(np.abs(phi[wrong_sign_mask]/phi_bound) > 1.e-4):
raise ValueError("Unphysical phi encountered when mapping to phi")
else:
phi[wrong_sign_mask] = 0.0
return phi
else:
raise NotImplementedError(f"Conversion to {destination} not yet implemented.")

Expand Down
Loading