use std::collections::BTreeMap;
use std::str::FromStr;
use pyo3::exceptions::{PyTypeError, PyValueError};
use pyo3::prelude::*;
use numpy::{IntoPyArray, PyArray1, PyArray2, PyArrayMethods};
use nucleide_nuclei::NuclideId;
#[pyfunction]
fn version() -> &'static str {
env!("CARGO_PKG_VERSION")
}
fn wrap_nucid_err(e: nucleide_nuclei::Error) -> PyErr {
PyValueError::new_err(e.to_string())
}
#[pyclass(name = "Nuclide")]
struct PyNuclide {
inner: NuclideId,
}
#[pymethods]
impl PyNuclide {
#[new]
fn new(name: &str) -> PyResult<Self> {
NuclideId::from_name(name)
.map(|inner| Self { inner })
.map_err(wrap_nucid_err)
}
#[getter]
fn name(&self) -> String {
self.inner.to_name()
}
#[getter]
fn nucid(&self) -> u32 {
self.inner.nucid()
}
#[getter]
fn zzaaam(&self) -> u32 {
self.inner.zzaaam()
}
#[getter]
fn z(&self) -> u32 {
self.inner.z()
}
#[getter]
fn a(&self) -> u32 {
self.inner.a()
}
#[getter]
fn state(&self) -> u32 {
self.inner.state()
}
#[getter]
fn zaid(&self) -> u32 {
nucleide_nuclei::dialects::to_zaid(self.inner)
}
#[getter]
fn zzllaaam(&self) -> String {
nucleide_nuclei::dialects::zzllaaam(self.inner)
}
#[getter]
fn serpent(&self) -> String {
nucleide_nuclei::dialects::serpent(self.inner)
}
#[getter]
fn nist(&self) -> String {
nucleide_nuclei::dialects::nist(self.inner)
}
#[getter]
fn cinder(&self) -> u32 {
nucleide_nuclei::dialects::to_cinder(self.inner)
}
#[getter]
fn alara(&self) -> String {
nucleide_nuclei::dialects::alara(self.inner)
}
#[getter]
fn sza(&self) -> u32 {
nucleide_nuclei::dialects::to_sza(self.inner)
}
fn fluka(&self) -> PyResult<&'static str> {
nucleide_nuclei::dialects::id_to_fluka(self.inner)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[getter]
fn mass(&self) -> Option<f64> {
nucleide_nuclei::data::atomic_mass(self.inner.nucid())
}
#[getter]
fn abundance(&self) -> Option<f64> {
nucleide_nuclei::data::natural_abundance(self.inner.nucid())
}
fn __repr__(&self) -> String {
format!("Nuclide({})", self.inner.to_name())
}
}
#[pyfunction]
fn from_zaid(zaid: u32) -> PyResult<PyNuclide> {
nucleide_nuclei::dialects::from_zaid(zaid)
.map(|inner| PyNuclide { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn lookup(key: &Bound<'_, PyAny>, f: impl Fn(u32) -> Option<f64>) -> PyResult<Option<f64>> {
if let Ok(nucid) = key.extract::<u32>() {
return Ok(f(nucid));
}
if let Ok(name) = key.extract::<&str>() {
let id = NuclideId::from_name(name).map_err(wrap_nucid_err)?;
return Ok(f(id.nucid()));
}
Err(PyTypeError::new_err("expected int nucid or str name"))
}
#[pyfunction]
fn atomic_mass(key: &Bound<'_, PyAny>) -> PyResult<Option<f64>> {
lookup(key, nucleide_nuclei::data::atomic_mass)
}
#[pyfunction]
fn natural_abundance(key: &Bound<'_, PyAny>) -> PyResult<Option<f64>> {
lookup(key, nucleide_nuclei::data::natural_abundance)
}
#[pyclass(name = "Particle")]
struct PyParticle {
inner: nucleide_nuclei::particles::ParticleId,
}
#[pymethods]
impl PyParticle {
#[new]
fn new(spec: &Bound<'_, PyAny>) -> PyResult<Self> {
let inner = if let Ok(pdc) = spec.extract::<i32>() {
nucleide_nuclei::particles::ParticleId::from_pdc(pdc)
.ok_or_else(|| PyValueError::new_err(format!("unknown PDC code {pdc}")))?
} else if let Ok(s) = spec.extract::<&str>() {
s.parse::<nucleide_nuclei::particles::ParticleId>()
.map_err(|e| PyValueError::new_err(e.to_string()))?
} else {
return Err(PyTypeError::new_err("expected str alias or int PDC"));
};
Ok(Self { inner })
}
#[getter]
fn name(&self) -> &'static str {
self.inner.name()
}
#[getter]
fn describe(&self) -> &'static str {
self.inner.describe()
}
fn mcnp(&self) -> Option<&'static str> {
self.inner.mcnp()
}
fn mcnp6(&self) -> Option<&'static str> {
self.inner.mcnp6()
}
fn fluka(&self) -> Option<&'static str> {
self.inner.fluka()
}
fn geant4(&self) -> Option<&'static str> {
self.inner.geant4()
}
fn __repr__(&self) -> String {
format!("Particle('{}')", self.inner.name())
}
}
#[pyfunction]
fn rxname_id(name: &str) -> PyResult<u32> {
nucleide_nuclei::rxname::name_to_id(name).map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn rxname_name(id: u32) -> Option<&'static str> {
nucleide_nuclei::rxname::id_to_name(id)
}
#[pyfunction]
fn rxname_mt(id: u32) -> i32 {
nucleide_nuclei::rxname::id_to_mt(id)
}
fn io_err(e: nucleide_mcnp_io::xsdir::Error) -> PyErr {
PyValueError::new_err(e.to_string())
}
fn m_err<T>(r: Result<T, impl std::fmt::Display>) -> PyResult<T> {
r.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyclass(name = "XsdirTable")]
struct PyXsdirTable {
inner: nucleide_mcnp_io::xsdir::XsdirTable,
}
#[pymethods]
impl PyXsdirTable {
#[getter]
fn name(&self) -> &str {
&self.inner.name
}
#[getter]
fn awr(&self) -> f64 {
self.inner.awr
}
#[getter]
fn filename(&self) -> &str {
&self.inner.filename
}
#[getter]
fn filetype(&self) -> i64 {
self.inner.filetype
}
#[getter]
fn address(&self) -> i64 {
self.inner.address
}
#[getter]
fn tablelength(&self) -> i64 {
self.inner.tablelength
}
#[getter]
fn temperature(&self) -> Option<f64> {
self.inner.temperature
}
#[getter]
fn ptable(&self) -> bool {
self.inner.ptable
}
fn zaid(&self) -> &str {
self.inner.zaid()
}
fn to_serpent(&self, directory: &str) -> PyResult<String> {
m_err(self.inner.to_serpent(directory))
}
fn __repr__(&self) -> String {
format!("<XsdirTable: {}>", self.inner.name)
}
}
#[pyclass(name = "Xsdir")]
struct PyXsdir {
inner: nucleide_mcnp_io::xsdir::Xsdir,
}
#[pymethods]
impl PyXsdir {
#[getter]
fn datapath(&self) -> Option<&str> {
self.inner.datapath.as_deref()
}
#[getter]
fn awr(&self) -> BTreeMap<u32, f64> {
self.inner.awr.clone()
}
#[getter]
fn tables(&self) -> Vec<PyXsdirTable> {
self.inner
.tables
.iter()
.map(|t| PyXsdirTable { inner: t.clone() })
.collect()
}
fn find_table(&self, name: &str) -> Vec<PyXsdirTable> {
self.inner
.find_table(name)
.into_iter()
.map(|t| PyXsdirTable { inner: t.clone() })
.collect()
}
fn nucs(&self) -> Vec<u32> {
self.inner.nucs().iter().map(|n| n.nucid()).collect()
}
}
#[pyfunction]
fn read_xsdir(path: &str) -> PyResult<PyXsdir> {
nucleide_mcnp_io::xsdir::Xsdir::from_file(path)
.map(|inner| PyXsdir { inner })
.map_err(io_err)
}
#[pyclass(name = "MeshTally")]
struct PyMeshTally {
inner: nucleide_mcnp_io::meshtal::MeshTallyData,
}
#[pymethods]
impl PyMeshTally {
#[getter]
fn tally_number(&self) -> u32 {
self.inner.tally_number
}
#[getter]
fn particle(&self) -> char {
self.inner.particle.letter()
}
#[getter]
fn dose_response(&self) -> bool {
self.inner.dose_response
}
#[getter]
fn x_bounds(&self) -> Vec<f64> {
self.inner.x_bounds.clone()
}
#[getter]
fn y_bounds(&self) -> Vec<f64> {
self.inner.y_bounds.clone()
}
#[getter]
fn z_bounds(&self) -> Vec<f64> {
self.inner.z_bounds.clone()
}
#[getter]
fn e_bounds(&self) -> Vec<f64> {
self.inner.e_bounds.clone()
}
fn dims(&self) -> [usize; 3] {
self.inner.dims()
}
fn num_ves(&self) -> usize {
self.inner.num_ves()
}
fn num_e_groups(&self) -> usize {
self.inner.num_e_groups()
}
fn cell(&self, i: usize, j: usize, k: usize) -> (Vec<f64>, Vec<f64>) {
let (r, e) = self.inner.cell(i, j, k);
(r.to_vec(), e.to_vec())
}
fn cell_total(&self, i: usize, j: usize, k: usize) -> (f64, f64) {
self.inner.cell_total(i, j, k)
}
#[getter]
fn result(&self) -> Vec<Vec<f64>> {
self.inner.result.clone()
}
#[getter]
fn rel_error(&self) -> Vec<Vec<f64>> {
self.inner.rel_error.clone()
}
#[getter]
fn total_result(&self) -> Vec<f64> {
self.inner.total_result.clone()
}
#[getter]
fn total_rel_error(&self) -> Vec<f64> {
self.inner.total_rel_error.clone()
}
fn to_list(&self) -> (Vec<Vec<f64>>, Vec<Vec<f64>>) {
(self.inner.result.clone(), self.inner.rel_error.clone())
}
fn totals_list(&self) -> (Vec<f64>, Vec<f64>) {
(
self.inner.total_result.clone(),
self.inner.total_rel_error.clone(),
)
}
#[allow(clippy::type_complexity)]
fn result_array<'py>(
&self,
py: Python<'py>,
) -> PyResult<(Bound<'py, PyArray2<f64>>, Bound<'py, PyArray2<f64>>)> {
let n_ve = self.inner.num_ves();
let n_g = self.inner.num_e_groups();
let flatten = |rows: &[Vec<f64>], name: &str| -> PyResult<Vec<f64>> {
if rows.len() != n_ve {
return Err(PyValueError::new_err(format!(
"tally {name}: expected {n_ve} rows, found {}",
rows.len()
)));
}
let mut flat = Vec::with_capacity(n_ve * n_g);
for (ve, row) in rows.iter().enumerate() {
if row.len() != n_g {
return Err(PyValueError::new_err(format!(
"tally {name}: row {ve} has {} groups, expected {n_g}",
row.len()
)));
}
flat.extend_from_slice(row);
}
Ok(flat)
};
let flat_r = flatten(&self.inner.result, "result")?;
let flat_e = flatten(&self.inner.rel_error, "rel_error")?;
let arr_r = m_err(
flat_r
.into_pyarray(py)
.reshape((n_ve, n_g))
.map_err(|e| e.to_string()),
)?;
let arr_e = m_err(
flat_e
.into_pyarray(py)
.reshape((n_ve, n_g))
.map_err(|e| e.to_string()),
)?;
Ok((arr_r, arr_e))
}
#[allow(clippy::type_complexity)]
fn totals_array<'py>(
&self,
py: Python<'py>,
) -> PyResult<(Bound<'py, PyArray1<f64>>, Bound<'py, PyArray1<f64>>)> {
Ok((
self.inner.total_result.clone().into_pyarray(py),
self.inner.total_rel_error.clone().into_pyarray(py),
))
}
}
#[pyclass(name = "Meshtal")]
struct PyMeshtal {
inner: nucleide_mcnp_io::meshtal::Meshtal,
}
#[pymethods]
impl PyMeshtal {
#[getter]
fn version(&self) -> &str {
&self.inner.version
}
#[getter]
fn ld(&self) -> &str {
&self.inner.ld
}
#[getter]
fn title(&self) -> &str {
&self.inner.title
}
#[getter]
fn histories(&self) -> u64 {
self.inner.histories
}
#[getter]
fn tallies(&self) -> BTreeMap<u32, PyMeshTally> {
self.inner
.tallies
.iter()
.map(|(k, v)| (*k, PyMeshTally { inner: v.clone() }))
.collect()
}
}
#[pyfunction]
fn read_meshtal(path: &str) -> PyResult<PyMeshtal> {
m_err(nucleide_mcnp_io::meshtal::Meshtal::from_file(path).map(|inner| PyMeshtal { inner }))
}
#[pyclass(name = "Wwinp")]
struct PyWwinp {
inner: nucleide_mcnp_io::wwinp::Wwinp,
}
#[pymethods]
impl PyWwinp {
#[getter]
fn ni(&self) -> u32 {
self.inner.ni
}
#[getter]
fn nr(&self) -> u32 {
self.inner.nr
}
#[getter]
fn ne(&self) -> Vec<u32> {
self.inner.ne.clone()
}
#[getter]
fn nf(&self) -> [u32; 3] {
self.inner.nf
}
#[getter]
fn origin(&self) -> [f64; 3] {
self.inner.origin
}
#[getter]
fn nc(&self) -> [u32; 3] {
self.inner.nc
}
#[getter]
fn cm(&self) -> Vec<Vec<f64>> {
self.inner.cm.clone()
}
#[getter]
fn bounds(&self) -> Vec<Vec<f64>> {
self.inner.bounds.clone()
}
#[getter]
fn e(&self) -> Vec<Vec<f64>> {
self.inner.e.clone()
}
fn ww_row(&self, particle: usize, group: usize) -> Vec<f64> {
self.inner.ww[particle][group].clone()
}
fn ww_column(&self, particle: usize, ve: usize) -> Vec<f64> {
self.inner.ww_column(particle, ve)
}
fn ww_row_array<'py>(
&self,
py: Python<'py>,
particle: usize,
group: usize,
) -> PyResult<Bound<'py, PyArray1<f64>>> {
let row = self
.inner
.ww
.get(particle)
.and_then(|groups| groups.get(group))
.ok_or_else(|| {
PyValueError::new_err(format!(
"ww_row_array: particle {particle} group {group} out of range"
))
})?;
Ok(row.clone().into_pyarray(py))
}
fn ww_column_array<'py>(
&self,
py: Python<'py>,
particle: usize,
ve: usize,
) -> PyResult<Bound<'py, PyArray1<f64>>> {
let groups = self.inner.ww.get(particle).ok_or_else(|| {
PyValueError::new_err(format!("ww_column_array: particle {particle} out of range"))
})?;
if groups.is_empty() {
return Err(PyValueError::new_err(format!(
"ww_column_array: particle {particle} has no groups"
)));
}
let nft = groups[0].len();
if ve >= nft {
return Err(PyValueError::new_err(format!(
"ww_column_array: ve {ve} out of range for {nft} volume elements"
)));
}
for (g, row) in groups.iter().enumerate() {
if row.len() != nft {
return Err(PyValueError::new_err(format!(
"ww particle {particle}: group {g} has {} values, expected {nft}",
row.len()
)));
}
}
let col: Vec<f64> = groups.iter().map(|row| row[ve]).collect();
Ok(col.into_pyarray(py))
}
fn ww_particle_array<'py>(
&self,
py: Python<'py>,
particle: usize,
) -> PyResult<Bound<'py, PyArray2<f64>>> {
let groups = self.inner.ww.get(particle).ok_or_else(|| {
PyValueError::new_err(format!(
"ww_particle_array: particle {particle} out of range"
))
})?;
if groups.is_empty() {
return Err(PyValueError::new_err(format!(
"ww_particle_array: particle {particle} has no groups"
)));
}
let nft = groups[0].len();
let mut flat = Vec::with_capacity(groups.len() * nft);
for (g, row) in groups.iter().enumerate() {
if row.len() != nft {
return Err(PyValueError::new_err(format!(
"ww particle {particle}: group {g} has {} values, expected {nft}",
row.len()
)));
}
flat.extend_from_slice(row);
}
let n_g = groups.len();
m_err(
flat.into_pyarray(py)
.reshape((n_g, nft))
.map_err(|e| e.to_string()),
)
}
}
#[pyfunction]
fn read_wwinp(path: &str) -> PyResult<PyWwinp> {
m_err(nucleide_mcnp_io::wwinp::Wwinp::from_file(path).map(|inner| PyWwinp { inner }))
}
fn mctal_card_dict<'py>(
py: Python<'py>,
card: &nucleide_mcnp_io::mctal::BinCard,
) -> PyResult<pyo3::Bound<'py, pyo3::types::PyDict>> {
let c = pyo3::types::PyDict::new(py);
c.set_item("count", card.count)?;
c.set_item("values", card.values.clone())?;
c.set_item("variant", card.variant.map(|v| v.to_string()))?;
c.set_item("flag", card.flag)?;
Ok(c)
}
#[pyclass(name = "Mctal")]
struct PyMctal {
inner: nucleide_mcnp_io::mctal::Mctal,
}
#[pymethods]
impl PyMctal {
#[getter]
fn code_name(&self) -> &str {
&self.inner.code_name
}
#[getter]
fn comment(&self) -> &str {
&self.inner.comment
}
#[getter]
fn n_histories(&self) -> u64 {
self.inner.n_histories
}
#[getter]
fn n_cycles(&self) -> usize {
self.inner.n_cycles
}
#[getter]
fn n_inactive(&self) -> usize {
self.inner.n_inactive
}
#[getter]
fn vars_per_cycle(&self) -> usize {
self.inner.vars_per_cycle
}
#[getter]
fn k_col(&self) -> Vec<f64> {
self.inner.k_col.clone()
}
#[getter]
fn k_abs(&self) -> Vec<f64> {
self.inner.k_abs.clone()
}
#[getter]
fn k_path(&self) -> Vec<f64> {
self.inner.k_path.clone()
}
#[getter]
fn prompt_life_col(&self) -> Vec<f64> {
self.inner.prompt_life_col.clone()
}
#[getter]
fn prompt_life_path(&self) -> Vec<f64> {
self.inner.prompt_life_path.clone()
}
#[getter]
fn averages(&self) -> Vec<BTreeMap<String, f64>> {
self.inner
.averages
.iter()
.map(|a| {
let mut m = BTreeMap::new();
m.insert("avg_k_col".into(), a.avg_k_col.0);
m.insert("avg_k_col_stdev".into(), a.avg_k_col.1);
m.insert("avg_k_abs".into(), a.avg_k_abs.0);
m.insert("avg_k_abs_stdev".into(), a.avg_k_abs.1);
m.insert("avg_k_path".into(), a.avg_k_path.0);
m.insert("avg_k_path_stdev".into(), a.avg_k_path.1);
m.insert("avg_k_combined".into(), a.avg_k_combined.0);
m.insert("avg_k_combined_stdev".into(), a.avg_k_combined.1);
m.insert("avg_k_combined_active".into(), a.avg_k_combined_active.0);
m.insert(
"avg_k_combined_active_stdev".into(),
a.avg_k_combined_active.1,
);
m.insert("prompt_life_combined".into(), a.prompt_life_combined.0);
m.insert(
"prompt_life_combined_stdev".into(),
a.prompt_life_combined.1,
);
m.insert("cycle_histories".into(), a.cycle_histories);
m.insert("fom".into(), a.fom);
m
})
.collect()
}
#[allow(clippy::type_complexity)]
fn k_arrays<'py>(
&self,
py: Python<'py>,
) -> PyResult<(
Bound<'py, PyArray1<f64>>,
Bound<'py, PyArray1<f64>>,
Bound<'py, PyArray1<f64>>,
Bound<'py, PyArray1<f64>>,
Bound<'py, PyArray1<f64>>,
)> {
Ok((
self.inner.k_col.clone().into_pyarray(py),
self.inner.k_abs.clone().into_pyarray(py),
self.inner.k_path.clone().into_pyarray(py),
self.inner.prompt_life_col.clone().into_pyarray(py),
self.inner.prompt_life_path.clone().into_pyarray(py),
))
}
fn averages_array<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyArray2<f64>>> {
let n = self.inner.averages.len();
let mut flat = Vec::with_capacity(n * 14);
for a in &self.inner.averages {
flat.extend_from_slice(&[
a.avg_k_col.0,
a.avg_k_col.1,
a.avg_k_abs.0,
a.avg_k_abs.1,
a.avg_k_path.0,
a.avg_k_path.1,
a.avg_k_combined.0,
a.avg_k_combined.1,
a.avg_k_combined_active.0,
a.avg_k_combined_active.1,
a.prompt_life_combined.0,
a.prompt_life_combined.1,
a.cycle_histories,
a.fom,
]);
}
m_err(
flat.into_pyarray(py)
.reshape((n, 14))
.map_err(|e| e.to_string()),
)
}
#[getter]
fn npert(&self) -> Option<String> {
self.inner.npert.clone()
}
#[getter]
fn tally_nums(&self) -> Vec<u32> {
self.inner.tally_nums.clone()
}
#[getter]
fn tallies(&self, py: Python<'_>) -> PyResult<Vec<Py<PyAny>>> {
use pyo3::types::PyDict;
let mut out = Vec::with_capacity(self.inner.tallies.len());
for t in &self.inner.tallies {
let d = PyDict::new(py);
d.set_item("number", t.number)?;
d.set_item("particle_type", t.particle_type)?;
d.set_item("detector_type", t.detector_type)?;
d.set_item("particle_list", t.particle_list.clone())?;
d.set_item("comment", t.comment.clone())?;
for (key, card) in [
("f", &t.f),
("d", &t.d),
("u", &t.u),
("s", &t.s),
("m", &t.m),
("c", &t.c),
("e", &t.e),
("t", &t.t),
] {
d.set_item(key, mctal_card_dict(py, card)?)?;
}
let vals: Vec<(f64, f64)> = t.vals.clone();
d.set_item("vals", vals)?;
let tfc_obj = if let Some(tfc) = &t.tfc {
let td = PyDict::new(py);
td.set_item("jtf", tfc.jtf.clone())?;
let mut rows = Vec::with_capacity(tfc.rows.len());
for r in &tfc.rows {
let rd = PyDict::new(py);
rd.set_item("nps", r.nps)?;
rd.set_item("value", r.value)?;
rd.set_item("rel_err", r.rel_err)?;
rd.set_item("fom", r.fom)?;
rows.push(rd.into_any().unbind());
}
td.set_item("rows", rows)?;
td.into_any().unbind()
} else {
py.None()
};
d.set_item("tfc", tfc_obj)?;
d.set_item("total", t.total_val())?;
out.push(d.into_any().unbind());
}
Ok(out)
}
#[getter]
fn mesh_tallies(&self, py: Python<'_>) -> PyResult<Vec<Py<PyAny>>> {
use pyo3::types::PyDict;
let mut out = Vec::with_capacity(self.inner.mesh_tallies.len());
for t in &self.inner.mesh_tallies {
let d = PyDict::new(py);
d.set_item("number", t.number)?;
d.set_item("particle_type", t.particle_type)?;
d.set_item("detector_type", t.detector_type)?;
d.set_item("particle_list", t.particle_list.clone())?;
d.set_item("comment", t.comment.clone())?;
d.set_item("mesh_unknown", t.mesh_unknown)?;
d.set_item("ni", t.ni)?;
d.set_item("nj", t.nj)?;
d.set_item("nk", t.nk)?;
d.set_item("dims", t.dims().to_vec())?;
d.set_item("num_cells", t.num_cells())?;
d.set_item("cora", t.cora.clone())?;
d.set_item("corb", t.corb.clone())?;
d.set_item("corc", t.corc.clone())?;
for (key, card) in [
("d", &t.d),
("u", &t.u),
("s", &t.s),
("m", &t.m),
("c", &t.c),
("e", &t.e),
("t", &t.t),
] {
d.set_item(key, mctal_card_dict(py, card)?)?;
}
let vals: Vec<(f64, f64)> = t.vals.clone();
d.set_item("vals", vals)?;
d.set_item("total", t.total_val())?;
out.push(d.into_any().unbind());
}
Ok(out)
}
fn tally_vals_array<'py>(
&self,
py: Python<'py>,
number: u32,
) -> PyResult<Bound<'py, PyArray2<f64>>> {
let tally = self
.inner
.tallies
.iter()
.find(|t| t.number == number)
.ok_or_else(|| {
PyValueError::new_err(format!("mctal has no parsed body for tally {number}"))
})?;
let mut flat = Vec::with_capacity(tally.vals.len() * 2);
for (v, e) in &tally.vals {
flat.push(*v);
flat.push(*e);
}
let n = tally.vals.len();
m_err(
flat.into_pyarray(py)
.reshape((n, 2))
.map_err(|e| e.to_string()),
)
}
fn mesh_tally_vals_array<'py>(
&self,
py: Python<'py>,
number: u32,
) -> PyResult<Bound<'py, PyArray2<f64>>> {
let tally = self
.inner
.mesh_tallies
.iter()
.find(|t| t.number == number)
.ok_or_else(|| {
PyValueError::new_err(format!("mctal has no parsed mesh body for tally {number}"))
})?;
let mut flat = Vec::with_capacity(tally.vals.len() * 2);
for (v, e) in &tally.vals {
flat.push(*v);
flat.push(*e);
}
let n = tally.vals.len();
m_err(
flat.into_pyarray(py)
.reshape((n, 2))
.map_err(|e| e.to_string()),
)
}
}
#[pyfunction]
fn read_mctal(path: &str) -> PyResult<PyMctal> {
m_err(nucleide_mcnp_io::mctal::Mctal::from_file(path).map(|inner| PyMctal { inner }))
}
#[pyclass(name = "SurfSrc")]
struct PySurfSrc {
inner: nucleide_mcnp_io::surfsrc::SurfSrc,
}
#[pymethods]
impl PySurfSrc {
#[getter]
fn kod(&self) -> String {
self.inner.header.kod.trim_end().to_string()
}
#[getter]
fn ver(&self) -> String {
self.inner.header.ver.trim_end().to_string()
}
#[getter]
fn np1(&self) -> i64 {
self.inner.header.np1
}
#[getter]
fn orignp1(&self) -> i64 {
self.inner.header.orignp1
}
#[getter]
fn nrss(&self) -> i64 {
self.inner.header.nrss
}
#[getter]
fn ncrd(&self) -> i32 {
self.inner.header.ncrd
}
#[getter]
fn njsw(&self) -> i32 {
self.inner.header.njsw
}
#[getter]
fn niss(&self) -> i64 {
self.inner.header.niss
}
fn print_header(&self) -> String {
self.inner.header.print_header()
}
fn tracks(&self) -> PyResult<Vec<BTreeMap<String, f64>>> {
let tracks = self
.inner
.read_tracklist()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(tracks
.iter()
.map(|t| {
let mut d = BTreeMap::new();
d.insert("nps".into(), t.nps);
d.insert("bitarray".into(), t.bitarray);
d.insert("wgt".into(), t.wgt);
d.insert("erg".into(), t.erg);
d.insert("tme".into(), t.tme);
d.insert("x".into(), t.x);
d.insert("y".into(), t.y);
d.insert("z".into(), t.z);
d.insert("u".into(), t.u);
d.insert("v".into(), t.v);
d.insert("cs".into(), t.cs);
d.insert("w".into(), t.w);
d
})
.collect())
}
}
#[pyfunction]
fn read_ssw(path: &str) -> PyResult<PySurfSrc> {
nucleide_mcnp_io::surfsrc::SurfSrc::open(path)
.map(|inner| PySurfSrc { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyclass(name = "PtracFile")]
struct PyPtracFile {
inner: nucleide_mcnp_io::ptrac::PtracFile,
}
#[pymethods]
impl PyPtracFile {
#[getter]
fn problem_title(&self) -> &str {
&self.inner.problem_title
}
#[getter]
fn width_code(&self) -> u8 {
match self.inner.format {
nucleide_mcnp_io::ptrac::Format::I4LittleEndian => 0,
nucleide_mcnp_io::ptrac::Format::I8LittleEndian => 1,
}
}
#[getter]
fn variable_nums(&self) -> BTreeMap<String, usize> {
let v = &self.inner.variable_nums;
let mut m = BTreeMap::new();
m.insert("nps".into(), v.nps);
m.insert("src".into(), v.src);
m.insert("bnk".into(), v.bnk);
m.insert("sur".into(), v.sur);
m.insert("col".into(), v.col);
m.insert("ter".into(), v.ter);
m
}
fn events(&self) -> PyResult<Vec<BTreeMap<String, f64>>> {
let events = self
.inner
.events()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(events
.iter()
.map(|ev| {
let mut d = BTreeMap::new();
d.insert("event_type".to_string(), ev.event_type as f64);
for (n, v) in ev.iter() {
d.insert(n.to_string(), v);
}
d
})
.collect())
}
fn events_array<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyArray2<f64>>> {
let events = self
.inner
.events()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let n = events.len();
let mut flat = Vec::with_capacity(n * PTRAC_EVENT_COLUMNS.len());
for ev in &events {
flat.push(ev.event_type as f64);
for col in &PTRAC_EVENT_COLUMNS[1..] {
flat.push(ev.get(col).unwrap_or(0.0));
}
}
m_err(
flat.into_pyarray(py)
.reshape((n, PTRAC_EVENT_COLUMNS.len()))
.map_err(|e| e.to_string()),
)
}
fn event_field_array<'py>(
&self,
py: Python<'py>,
field: &str,
) -> PyResult<Bound<'py, PyArray1<f64>>> {
if !PTRAC_EVENT_COLUMNS.contains(&field) {
return Err(PyValueError::new_err(format!(
"unknown PTRAC field `{field}` (expected one of {})",
PTRAC_EVENT_COLUMNS.join(", ")
)));
}
let events = self
.inner
.events()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let col: Vec<f64> = events
.iter()
.map(|ev| {
if field == "event_type" {
ev.event_type as f64
} else {
ev.get(field).unwrap_or(0.0)
}
})
.collect();
Ok(col.into_pyarray(py))
}
}
const PTRAC_EVENT_COLUMNS: [&str; 19] = [
"event_type",
"node",
"nsr",
"nsf",
"nxs",
"ntyn",
"ipt",
"ncl",
"mat",
"ncp",
"xxx",
"yyy",
"zzz",
"uuu",
"vvv",
"www",
"erg",
"wgt",
"tme",
];
#[pyfunction]
fn read_ptrac(path: &str) -> PyResult<PyPtracFile> {
nucleide_mcnp_io::ptrac::PtracFile::open(path)
.map(|inner| PyPtracFile { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn mcpl_particle_to_dict(py: Python<'_>, p: &nucleide_mcpl_io::Particle) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let d = PyDict::new(py);
d.set_item("ekin", p.ekin)?;
d.set_item("polarisation", p.polarisation.to_vec())?;
d.set_item("position", p.position.to_vec())?;
d.set_item("direction", p.direction.to_vec())?;
d.set_item("time", p.time)?;
d.set_item("weight", p.weight)?;
d.set_item("pdgcode", p.pdgcode)?;
d.set_item("userflags", p.userflags)?;
Ok(d.into_any().unbind())
}
fn mcpl_particle_from_dict(d: &Bound<'_, PyAny>) -> PyResult<nucleide_mcpl_io::Particle> {
let get_f64 = |key: &str| -> PyResult<f64> {
d.get_item(key)
.map_err(|e| PyValueError::new_err(format!("particle missing `{key}`: {e}")))?
.extract()
.map_err(|_| PyValueError::new_err(format!("particle `{key}` must be a float")))
};
let get_vec3 = |key: &str| -> PyResult<[f64; 3]> {
let v: Vec<f64> = d
.get_item(key)
.map_err(|e| PyValueError::new_err(format!("particle missing `{key}`: {e}")))?
.extract()
.map_err(|_| PyValueError::new_err(format!("particle `{key}` must be a 3-list")))?;
if v.len() != 3 {
return Err(PyValueError::new_err(format!(
"particle `{key}` must have exactly 3 entries"
)));
}
Ok([v[0], v[1], v[2]])
};
let pdgcode: i32 = d
.get_item("pdgcode")
.map_err(|e| PyValueError::new_err(format!("particle missing `pdgcode`: {e}")))?
.extract()
.map_err(|_| PyValueError::new_err("particle `pdgcode` must be an int"))?;
let userflags: u32 = d
.get_item("userflags")
.map_err(|e| PyValueError::new_err(format!("particle missing `userflags`: {e}")))?
.extract()
.map_err(|_| PyValueError::new_err("particle `userflags` must be an int"))?;
Ok(nucleide_mcpl_io::Particle {
ekin: get_f64("ekin")?,
polarisation: get_vec3("polarisation")?,
position: get_vec3("position")?,
direction: get_vec3("direction")?,
time: get_f64("time")?,
weight: get_f64("weight")?,
pdgcode,
userflags,
})
}
#[pyclass(name = "McplFile")]
struct PyMcplFile {
inner: nucleide_mcpl_io::McplFile,
}
#[pymethods]
impl PyMcplFile {
#[getter]
fn version(&self) -> u16 {
self.inner.header.version
}
#[getter]
fn nparticles(&self) -> u64 {
self.inner.header.nparticles
}
#[getter]
fn srcname(&self) -> &str {
&self.inner.header.srcname
}
#[getter]
fn comments(&self) -> Vec<String> {
self.inner.header.comments.clone()
}
#[getter]
fn has_userflags(&self) -> bool {
self.inner.header.has_userflags
}
#[getter]
fn has_polarisation(&self) -> bool {
self.inner.header.has_polarisation
}
#[getter]
fn double_prec(&self) -> bool {
self.inner.header.double_prec
}
#[getter]
fn universal_pdgcode(&self) -> Option<i32> {
self.inner.header.universal_pdgcode
}
#[getter]
fn universal_weight(&self) -> Option<f64> {
self.inner.header.universal_weight
}
#[getter]
fn blobs(&self) -> Vec<(String, Vec<u8>)> {
self.inner
.header
.blobs
.iter()
.map(|b| (b.key.clone(), b.data.clone()))
.collect()
}
fn particles(&self, py: Python<'_>) -> PyResult<Vec<Py<PyAny>>> {
let ps = self
.inner
.particles()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
ps.iter().map(|p| mcpl_particle_to_dict(py, p)).collect()
}
}
#[pyfunction]
fn read_mcpl(path: &str) -> PyResult<PyMcplFile> {
nucleide_mcpl_io::McplFile::open(path)
.map(|inner| PyMcplFile { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn write_mcpl(
path: &str,
header: &Bound<'_, PyAny>,
particles: Vec<Bound<'_, PyAny>>,
) -> PyResult<()> {
use nucleide_mcpl_io::{Blob, Header};
let get = |key: &str| header.get_item(key);
let srcname: String = get("srcname")
.map_err(|_| PyValueError::new_err("header missing `srcname`"))?
.extract()
.map_err(|_| PyValueError::new_err("header `srcname` must be a str"))?;
let comments: Vec<String> = get("comments")
.map_err(|_| PyValueError::new_err("header missing `comments`"))?
.extract()
.map_err(|_| PyValueError::new_err("header `comments` must be a list of str"))?;
let flag = |key: &str| -> PyResult<bool> {
get(key)
.map_err(|_| PyValueError::new_err(format!("header missing `{key}`")))?
.extract()
.map_err(|_| PyValueError::new_err(format!("header `{key}` must be a bool")))
};
let universal_pdgcode: Option<i32> = get("universal_pdgcode")
.map_err(|_| PyValueError::new_err("header missing `universal_pdgcode`"))?
.extract()
.map_err(|_| PyValueError::new_err("header `universal_pdgcode` must be an int or None"))?;
let universal_weight: Option<f64> = get("universal_weight")
.map_err(|_| PyValueError::new_err("header missing `universal_weight`"))?
.extract()
.map_err(|_| PyValueError::new_err("header `universal_weight` must be a float or None"))?;
let blob_pairs: Vec<(String, Vec<u8>)> = get("blobs")
.map_err(|_| PyValueError::new_err("header missing `blobs`"))?
.extract()
.map_err(|_| {
PyValueError::new_err("header `blobs` must be a list of (key, bytes) pairs")
})?;
let h = Header {
has_userflags: flag("has_userflags")?,
has_polarisation: flag("has_polarisation")?,
double_prec: flag("double_prec")?,
universal_pdgcode,
universal_weight,
srcname,
comments,
blobs: blob_pairs
.into_iter()
.map(|(key, data)| Blob { key, data })
.collect(),
..Header::default()
};
let ps: Vec<nucleide_mcpl_io::Particle> = particles
.iter()
.map(mcpl_particle_from_dict)
.collect::<PyResult<_>>()?;
nucleide_mcpl_io::write_to_path(path, &h, &ps).map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (ssw_path, mcpl_path, surfs, kinds, options=None))]
fn ssw2mcpl(
ssw_path: &str,
mcpl_path: &str,
surfs: Vec<u32>,
kinds: Vec<String>,
options: Option<Bound<'_, PyAny>>,
) -> PyResult<u64> {
use nucleide_mcnp_io::surfsrc::SurfSrc;
use nucleide_mcpl_io::ssw::{SswParticleKind, SswTrack};
let ssw = SurfSrc::open(ssw_path).map_err(|e| PyValueError::new_err(e.to_string()))?;
let raw = ssw
.read_tracklist()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
if raw.len() != surfs.len() || raw.len() != kinds.len() {
return Err(PyValueError::new_err(format!(
"ssw2mcpl: SSW holds {} tracks but got {} surfs and {} kinds \
(one surf+kind per track required)",
raw.len(),
surfs.len(),
kinds.len()
)));
}
let mut tracks = Vec::with_capacity(raw.len());
for (i, (t, surf, kind)) in raw
.iter()
.zip(surfs)
.zip(kinds.iter())
.map(|((t, s), k)| (t, s, k))
.enumerate()
{
let kind = SswParticleKind::parse(kind).ok_or_else(|| {
PyValueError::new_err(format!(
"track {i} kind `{kind}` unknown (expected one of \
\"neutron\", \"gamma\", \"electron\", \"positron\", \"proton\")"
))
})?;
tracks.push(SswTrack {
ekin: t.erg,
time_shakes: t.tme,
position: [t.x, t.y, t.z],
direction: [t.u, t.v, t.cs],
weight: t.wgt,
surf,
kind,
});
}
let mut opts = parse_ssw2mcpl_options(options.as_ref())?;
if mcpl_path.ends_with(".gz") {
opts.gzip = true;
}
let bytes = nucleide_mcpl_io::ssw::ssw2mcpl_bytes(&tracks, &opts)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
std::fs::write(mcpl_path, bytes).map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(tracks.len() as u64)
}
fn parse_ssw2mcpl_options(
options: Option<&Bound<'_, PyAny>>,
) -> PyResult<nucleide_mcpl_io::ssw::Ssw2McplOptions> {
use nucleide_mcpl_io::ssw::{DeckBlob, Ssw2McplOptions};
let mut opts = Ssw2McplOptions::default();
let Some(d) = options else {
return Ok(opts);
};
if !d.is_instance_of::<pyo3::types::PyDict>() {
return Err(PyValueError::new_err("options must be a dict or None"));
}
let flag = |key: &str| -> PyResult<Option<bool>> {
match d.get_item(key) {
Ok(v) => v
.extract()
.map(Some)
.map_err(|_| PyValueError::new_err(format!("options `{key}` must be a bool"))),
Err(_) => Ok(None),
}
};
if let Some(v) = flag("double_prec")? {
opts.double_prec = v;
}
if let Some(v) = flag("surf_to_userflags")? {
opts.surf_to_userflags = v;
}
if let Some(v) = flag("gzip")? {
opts.gzip = v;
}
if let Some(v) = flag("universal_pdg")? {
opts.universal_pdg = v;
}
if let Some(v) = flag("universal_weight")? {
opts.universal_weight = v;
}
if let Ok(v) = d.get_item("polarisation") {
if v.is_none() {
opts.polarisation = None;
} else {
let vec: Vec<f64> = v.extract().map_err(|_| {
PyValueError::new_err("options `polarisation` must be a 3-list or None")
})?;
if vec.len() != 3 {
return Err(PyValueError::new_err(
"options `polarisation` must have exactly 3 entries",
));
}
opts.polarisation = Some([vec[0], vec[1], vec[2]]);
}
}
if let Ok(v) = d.get_item("srcname") {
opts.srcname = v
.extract()
.map_err(|_| PyValueError::new_err("options `srcname` must be a str"))?;
}
if let Ok(v) = d.get_item("comments") {
opts.comments = v
.extract()
.map_err(|_| PyValueError::new_err("options `comments` must be a list of str"))?;
}
if let Ok(v) = d.get_item("deck_blob") {
if !v.is_none() {
let (key, data): (String, Vec<u8>) = v.extract().map_err(|_| {
PyValueError::new_err("options `deck_blob` must be a (key, bytes) pair or None")
})?;
opts.deck_blob = Some(DeckBlob { key, data });
}
}
Ok(opts)
}
#[pyfunction]
#[pyo3(signature = (mcpl_path, reference_ssw_path, ssw_out_path, surface=None, force_cs_to_one=false, niss=None, allow_polarisation=false))]
fn mcpl2ssw(
mcpl_path: &str,
reference_ssw_path: &str,
ssw_out_path: &str,
surface: Option<u32>,
force_cs_to_one: bool,
niss: Option<i64>,
allow_polarisation: bool,
) -> PyResult<u64> {
use nucleide_mcnp_io::surfsrc::SurfSrc;
use nucleide_mcpl_io::ssw::Mcpl2SswOptions;
let mcpl = nucleide_mcpl_io::McplFile::open(mcpl_path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let particles = mcpl
.particles()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let reference =
SurfSrc::open(reference_ssw_path).map_err(|e| PyValueError::new_err(e.to_string()))?;
let (header, tracks) = nucleide_mcpl_io::ssw::mcpl2ssw(
&particles,
&reference.header,
&Mcpl2SswOptions {
surface,
force_cs_to_one,
niss_override: niss,
allow_polarisation,
},
)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
nucleide_mcnp_io::surfsrc::write_to_path(ssw_out_path, &header, &tracks)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(tracks.len() as u64)
}
#[pyfunction]
fn merge_mcpl(paths: Vec<String>, out_path: &str) -> PyResult<u64> {
let mut files = Vec::with_capacity(paths.len());
for (i, p) in paths.iter().enumerate() {
files.push(
nucleide_mcpl_io::McplFile::open(p)
.map_err(|e| PyValueError::new_err(format!("merge_mcpl: input {i} {p}: {e}")))?,
);
}
let (header, particles) =
nucleide_mcpl_io::merge_mcpl(&files).map_err(|e| PyValueError::new_err(e.to_string()))?;
nucleide_mcpl_io::write_to_path(out_path, &header, &particles)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(particles.len() as u64)
}
struct ParsedExtract {
spec: nucleide_mcpl_io::ExtractSpec,
pending: Option<std::rc::Rc<std::cell::RefCell<Option<PyErr>>>>,
}
fn parse_extract_spec(
options: Option<&Bound<'_, PyAny>>,
nparticles: usize,
) -> PyResult<ParsedExtract> {
let Some(d) = options else {
return Ok(ParsedExtract {
spec: nucleide_mcpl_io::ExtractSpec::Range(0..nparticles),
pending: None,
});
};
if !d.is_instance_of::<pyo3::types::PyDict>() {
return Err(PyValueError::new_err("options must be a dict or None"));
}
let opt_usize = |key: &str| -> PyResult<Option<usize>> {
match d.get_item(key) {
Err(_) => Ok(None),
Ok(v) if v.is_none() => Ok(None),
Ok(v) => v.extract::<usize>().map(Some).map_err(|_| {
PyValueError::new_err(format!("options `{key}` must be a non-negative int"))
}),
}
};
let start = opt_usize("start")?;
let stop = opt_usize("stop")?;
if let Ok(cb) = d.get_item("predicate") {
if !cb.is_none() {
if start.is_some() || stop.is_some() {
return Err(PyValueError::new_err(
"options `start`/`stop` and `predicate` cannot be combined",
));
}
if !cb.is_callable() {
return Err(PyValueError::new_err(
"options `predicate` must be callable",
));
}
let cb = cb.unbind();
let pending: std::rc::Rc<std::cell::RefCell<Option<PyErr>>> =
std::rc::Rc::new(std::cell::RefCell::new(None));
let pending_inner = std::rc::Rc::clone(&pending);
let spec = nucleide_mcpl_io::ExtractSpec::Predicate(Box::new(
move |p: &nucleide_mcpl_io::Particle| -> bool {
if pending_inner.borrow().is_some() {
return false;
}
Python::attach(|py| {
let dict = match mcpl_particle_to_dict(py, p) {
Ok(d) => d,
Err(e) => {
*pending_inner.borrow_mut() = Some(e);
return false;
}
};
match cb.call1(py, (dict,)) {
Ok(v) => match v.is_truthy(py) {
Ok(t) => t,
Err(e) => {
*pending_inner.borrow_mut() = Some(e);
false
}
},
Err(e) => {
*pending_inner.borrow_mut() = Some(e);
false
}
}
})
},
));
return Ok(ParsedExtract {
spec,
pending: Some(pending),
});
}
}
Ok(ParsedExtract {
spec: nucleide_mcpl_io::ExtractSpec::Range(start.unwrap_or(0)..stop.unwrap_or(nparticles)),
pending: None,
})
}
#[pyfunction]
#[pyo3(signature = (src_path, out_path, options=None))]
fn extract_mcpl(
src_path: &str,
out_path: &str,
options: Option<Bound<'_, PyAny>>,
) -> PyResult<u64> {
let file = nucleide_mcpl_io::McplFile::open(src_path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let nparticles = file.header.nparticles as usize;
let parsed = parse_extract_spec(options.as_ref(), nparticles)?;
let (header, particles) = nucleide_mcpl_io::extract_mcpl(&file, &parsed.spec)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
if let Some(err) = parsed.pending.and_then(|p| p.borrow_mut().take()) {
return Err(err);
}
nucleide_mcpl_io::write_to_path(out_path, &header, &particles)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(particles.len() as u64)
}
#[pyfunction]
fn mcpl_stats(py: Python<'_>, path: &str) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let file =
nucleide_mcpl_io::McplFile::open(path).map_err(|e| PyValueError::new_err(e.to_string()))?;
let s =
nucleide_mcpl_io::mcpl_stats(&file).map_err(|e| PyValueError::new_err(e.to_string()))?;
let d = PyDict::new(py);
d.set_item("nparticles", s.nparticles)?;
d.set_item("ekin_sum", s.ekin_sum)?;
d.set_item("ekin_min", s.ekin_min)?;
d.set_item("ekin_max", s.ekin_max)?;
d.set_item("ekin_mean", s.ekin_mean)?;
d.set_item("weight_sum", s.weight_sum)?;
d.set_item("pdg_counts", s.pdg_counts)?;
Ok(d.into_any().unbind())
}
#[pyfunction]
fn repair_mcpl(path: &str) -> PyResult<u64> {
let file =
nucleide_mcpl_io::McplFile::open(path).map_err(|e| PyValueError::new_err(e.to_string()))?;
let repaired = nucleide_mcpl_io::repair_mcpl(&file);
let n = nucleide_mcpl_io::McplFile::from_bytes(repaired.clone())
.map_err(|e| PyValueError::new_err(e.to_string()))?
.header
.nparticles;
nucleide_mcpl_io::write_bytes_to_path(path, &repaired)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(n)
}
#[pyclass(name = "EndlLibrary")]
struct PyEndlLibrary {
inner: nucleide_mcnp_io::endl::Library,
}
#[pymethods]
impl PyEndlLibrary {
fn nuclides(&self) -> Vec<i64> {
self.inner.nuclides()
}
#[pyo3(signature = (nuc, p_in, rdesc, rprop, x1=None, p_out=None))]
fn get_rx(
&self,
nuc: &Bound<'_, PyAny>,
p_in: i32,
rdesc: i32,
rprop: i32,
x1: Option<i32>,
p_out: Option<i32>,
) -> PyResult<Vec<Vec<f64>>> {
let id = if let Ok(n) = nuc.extract::<i64>() {
n
} else if let Ok(name) = nuc.extract::<&str>() {
NuclideId::from_name(name).map_err(wrap_nucid_err)?.nucid() as i64
} else {
return Err(PyTypeError::new_err("expected int nucleus id or str name"));
};
self.inner
.get_rx(id, p_in, rdesc, rprop, x1, p_out)
.map(|rows| rows.to_vec())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
}
#[pyfunction]
fn read_endl(path: &str) -> PyResult<PyEndlLibrary> {
nucleide_mcnp_io::endl::Library::open(path)
.map(|inner| PyEndlLibrary { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn endl_endftod(field: &str) -> f64 {
nucleide_mcnp_io::endl::endftod(field)
}
#[pyfunction]
fn combine_ssw_files(output: &str, inputs: Vec<String>) -> PyResult<()> {
nucleide_mcnp_io::surfsrc::combine_files(&inputs, output)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyclass(name = "Chain")]
struct PyChain {
inner: std::sync::Arc<nucleide_depletion::Chain>,
}
#[pymethods]
impl PyChain {
#[getter]
fn nuclides(&self) -> Vec<String> {
self.inner.nuclides.iter().map(|n| n.name.clone()).collect()
}
fn index_of(&self, name: &str) -> Option<usize> {
self.inner.index_of(name)
}
}
#[pyfunction]
fn read_chain(path: &str) -> PyResult<PyChain> {
nucleide_depletion::Chain::from_file(path)
.map(|inner| PyChain {
inner: std::sync::Arc::new(inner),
})
.map_err(|e| PyValueError::new_err(e.to_string()))
}
type RateMap = BTreeMap<String, f64>;
#[pyclass(name = "DepletionSystem")]
struct PyDepletionSystem {
inner: std::sync::Arc<nucleide_depletion::DepletionSystem>,
}
#[pymethods]
impl PyDepletionSystem {
#[pyo3(signature = (n0, dt, order=48, method="cram48"))]
fn solve(
&self,
n0: BTreeMap<String, f64>,
dt: f64,
order: u8,
method: &str,
) -> PyResult<BTreeMap<String, f64>> {
let method = resolve_method(order, method)?;
nucleide_depletion::deplete_with_method(&self.inner, method, &n0, dt)
.map(|r| r.atoms)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyo3(signature = (n0, dt, order=48, method="cram48"))]
fn solve_vec(&self, n0: Vec<f64>, dt: f64, order: u8, method: &str) -> PyResult<Vec<f64>> {
let method = resolve_method(order, method)?;
nucleide_depletion::solve_with_method(&self.inner, method, &n0, dt)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
}
#[pyfunction]
fn build_depletion_system(chain: &PyChain, rates: RateMap) -> PyResult<PyDepletionSystem> {
let rs = split_rates(&rates, &chain.inner)?;
nucleide_depletion::DepletionSystem::build((*chain.inner).clone(), &rs)
.map(|sys| PyDepletionSystem {
inner: std::sync::Arc::new(sys),
})
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn parse_order(order: u8) -> PyResult<nucleide_depletion::Order> {
match order {
16 => Ok(nucleide_depletion::Order::Order16),
48 => Ok(nucleide_depletion::Order::Order48),
other => Err(PyValueError::new_err(format!(
"unsupported CRAM order {other} (supported: 16, 48)"
))),
}
}
fn parse_method(name: &str) -> PyResult<nucleide_depletion::Method> {
name.parse().map_err(|e: String| PyValueError::new_err(e))
}
fn resolve_method(order: u8, method: &str) -> PyResult<nucleide_depletion::Method> {
let parsed = parse_method(method)?;
if parsed == nucleide_depletion::Method::default_cram() {
parse_order(order).map(nucleide_depletion::Method::Cram)
} else {
Ok(parsed)
}
}
fn split_rates(
rates: &RateMap,
chain: &nucleide_depletion::Chain,
) -> PyResult<nucleide_depletion::ReactionRates> {
let mut out = nucleide_depletion::ReactionRates::new();
for (key, v) in rates {
let (nuc, rx) = key.split_once(':').ok_or_else(|| {
PyValueError::new_err(format!("rate key `{key}` must be `Name:reaction`"))
})?;
let idx = chain
.index_of(nuc)
.ok_or_else(|| PyValueError::new_err(format!("rate for unknown nuclide `{nuc}`")))?;
out.entry(idx).or_default().insert(rx.to_string(), *v);
}
Ok(out)
}
#[pyfunction]
#[pyo3(signature = (chain, n0, dt, rates=None, order=48, method="cram48"))]
fn deplete(
chain: &PyChain,
n0: BTreeMap<String, f64>,
dt: f64,
rates: Option<RateMap>,
order: u8,
method: &str,
) -> PyResult<BTreeMap<String, f64>> {
let method = resolve_method(order, method)?;
let rates = split_rates(rates.as_ref().unwrap_or(&BTreeMap::new()), &chain.inner)?;
let sys = nucleide_depletion::DepletionSystem::build((*chain.inner).clone(), &rates)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
nucleide_depletion::deplete_with_method(&sys, method, &n0, dt)
.map(|r| r.atoms)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn read_serpent(path: &str, kind: &str) -> PyResult<Py<PyAny>> {
let text = std::fs::read_to_string(path).map_err(|e| PyValueError::new_err(e.to_string()))?;
let table = match kind {
"res" => nucleide_serpent_io::parse_res(&text),
"dep" => nucleide_serpent_io::parse_dep(&text),
"det" => nucleide_serpent_io::parse_det(&text),
other => {
return Err(PyValueError::new_err(format!(
"kind must be res|dep|det, got `{other}`"
)))
}
}
.map_err(|e| PyValueError::new_err(e.to_string()))?;
fn entry_to_py(py: Python<'_>, e: &nucleide_serpent_io::Entry) -> PyResult<Py<PyAny>> {
use nucleide_serpent_io::Entry as E;
let value = match e {
E::Scalar(nucleide_serpent_io::Value::Num(n)) => {
n.into_pyobject(py).unwrap().unbind().into_any()
}
E::Scalar(nucleide_serpent_io::Value::Str(s)) => {
s.into_pyobject(py).unwrap().unbind().into_any()
}
E::Vector(vs) => vs
.iter()
.map(|v| match v {
nucleide_serpent_io::Value::Num(n) => {
n.into_pyobject(py).unwrap().unbind().into_any()
}
nucleide_serpent_io::Value::Str(s) => {
s.into_pyobject(py).unwrap().unbind().into_any()
}
})
.collect::<Vec<_>>()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
E::Matrix(m) => m
.to_rows_f64()
.map_err(|err| PyValueError::new_err(err.to_string()))?
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
};
Ok(value)
}
Python::attach(|py| {
let dict = pyo3::types::PyDict::new(py);
for (k, e) in table.iter() {
dict.set_item(k, entry_to_py(py, e)?)?;
}
Ok(dict.into_any().unbind())
})
}
#[pyclass(name = "UsrbinTally")]
struct PyUsrbinTally {
inner: nucleide_fluka_io::usrbin::UsrbinTally,
}
#[pymethods]
impl PyUsrbinTally {
#[getter]
fn name(&self) -> &str {
&self.inner.name
}
#[getter]
fn particle(&self) -> &str {
&self.inner.particle
}
#[getter]
fn nx(&self) -> usize {
self.inner.x_info.bins
}
#[getter]
fn ny(&self) -> usize {
self.inner.y_info.bins
}
#[getter]
fn nz(&self) -> usize {
self.inner.z_info.bins
}
#[getter]
fn x_bounds(&self) -> Vec<f64> {
self.inner.x_bounds.clone()
}
#[getter]
fn y_bounds(&self) -> Vec<f64> {
self.inner.y_bounds.clone()
}
#[getter]
fn z_bounds(&self) -> Vec<f64> {
self.inner.z_bounds.clone()
}
#[getter]
fn data(&self) -> Vec<f64> {
self.inner.part_data.clone()
}
#[getter]
fn error(&self) -> Vec<f64> {
self.inner.error_data.clone()
}
fn dims(&self) -> [usize; 3] {
[self.nx(), self.ny(), self.nz()]
}
}
#[pyfunction]
fn read_usrbin(path: &str) -> PyResult<Vec<PyUsrbinTally>> {
let tallies = nucleide_fluka_io::usrbin::read_usrbin_file(path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(tallies
.into_iter()
.map(|inner| PyUsrbinTally { inner })
.collect())
}
#[pyclass(name = "MagicOutput")]
struct PyMagicOutput {
inner: nucleide_vr_tools::magic::MagicOutput,
}
#[pymethods]
impl PyMagicOutput {
#[getter]
fn lower_bounds_ww(&self) -> Vec<f64> {
self.inner.lower_bounds_ww.clone()
}
#[getter]
fn groups_per_ve(&self) -> usize {
self.inner.groups_per_ve
}
#[getter]
fn scale_factors(&self) -> Vec<f64> {
self.inner.scale_factors.clone()
}
#[getter]
fn e_upper_bounds(&self) -> Vec<f64> {
self.inner.e_upper_bounds.clone()
}
#[getter]
fn ww_tag_name(&self) -> &str {
&self.inner.ww_tag_name
}
}
#[pyfunction]
#[pyo3(signature = (tally, per_group=false, tolerance=0.5))]
fn magic(tally: &PyMeshTally, per_group: bool, tolerance: f64) -> PyResult<PyMagicOutput> {
let selection = if per_group {
nucleide_vr_tools::magic::MagicSelection::PerGroup
} else {
nucleide_vr_tools::magic::MagicSelection::Total
};
let params = nucleide_vr_tools::magic::MagicParams {
tolerance,
..Default::default()
};
nucleide_vr_tools::magic::magic_with(&tally.inner, selection, params)
.map(|inner| PyMagicOutput { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyclass(name = "AliasTable")]
struct PyAliasTable {
inner: nucleide_vr_tools::sampling::AliasTable,
}
#[pymethods]
impl PyAliasTable {
#[new]
fn new(pdf: Vec<f64>) -> PyResult<Self> {
nucleide_vr_tools::sampling::AliasTable::new(&pdf)
.map(|inner| PyAliasTable { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn sample(&self, r1: f64, r2: f64) -> usize {
self.inner.sample(r1, r2)
}
#[getter]
fn pdf(&self) -> Vec<f64> {
self.inner.pdf().to_vec()
}
fn __len__(&self) -> usize {
self.inner.len()
}
}
#[pyclass(name = "MeshSourceSampler")]
struct PyMeshSourceSampler {
inner: nucleide_vr_tools::sampling::MeshSourceSampler,
}
#[pymethods]
impl PyMeshSourceSampler {
#[new]
#[pyo3(signature = (tally, mode, user_pdf=None))]
fn new(tally: &PyMeshTally, mode: &str, user_pdf: Option<Vec<f64>>) -> PyResult<Self> {
let user = if matches!(mode, "user") {
Some(user_pdf.ok_or_else(|| PyValueError::new_err("user mode needs user_pdf"))?)
} else {
None
};
let m = match mode {
"analog" => nucleide_vr_tools::sampling::Mode::Analog,
"uniform" => nucleide_vr_tools::sampling::Mode::Uniform,
"user" => nucleide_vr_tools::sampling::Mode::User,
other => {
return Err(PyValueError::new_err(format!(
"mode must be analog|uniform|user, got `{other}`"
)))
}
};
nucleide_vr_tools::sampling::MeshSourceSampler::new(&tally.inner, m, user.as_deref())
.map(|inner| PyMeshSourceSampler { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn sample(&self, r1: f64, r2: f64) -> BTreeMap<String, f64> {
let s = self.inner.sample(r1, r2);
let mut d = BTreeMap::new();
d.insert("index".into(), s.index as f64);
d.insert("i".into(), s.i as f64);
d.insert("j".into(), s.j as f64);
d.insert("k".into(), s.k as f64);
d.insert("weight".into(), s.weight);
d
}
fn mode(&self) -> &'static str {
match self.inner.mode() {
nucleide_vr_tools::sampling::Mode::Analog => "analog",
nucleide_vr_tools::sampling::Mode::Uniform => "uniform",
nucleide_vr_tools::sampling::Mode::User => "user",
}
}
fn num_voxels(&self) -> usize {
self.inner.num_voxels()
}
fn table_len(&self) -> usize {
self.inner.table().len()
}
}
#[pyclass(name = "KdeSampler")]
struct PyKdeSampler {
inner: nucleide_vr_tools::kde::KdeSampler,
}
#[pymethods]
impl PyKdeSampler {
#[new]
#[pyo3(signature = (samples, bandwidth=None))]
fn new(samples: Vec<Vec<f64>>, bandwidth: Option<&Bound<'_, PyAny>>) -> PyResult<Self> {
let rule = match bandwidth {
None => nucleide_vr_tools::kde::Bandwidth::Silverman,
Some(b) => {
if let Ok(name) = b.extract::<String>() {
match name.as_str() {
"silverman" => nucleide_vr_tools::kde::Bandwidth::Silverman,
other => {
return Err(PyValueError::new_err(format!(
"bandwidth must be silverman or a width list, got `{other}`"
)))
}
}
} else {
let widths = b.extract::<Vec<f64>>().map_err(|_| {
PyValueError::new_err("bandwidth must be silverman or a width list")
})?;
nucleide_vr_tools::kde::Bandwidth::Fixed(widths)
}
}
};
nucleide_vr_tools::kde::KdeSampler::fit(&samples, rule)
.map(|inner| PyKdeSampler { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn pdf(&self, point: Vec<f64>) -> PyResult<f64> {
self.inner
.pdf(&point)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn draw(&self, u: f64, normals: Vec<f64>) -> PyResult<Vec<f64>> {
self.inner
.draw(u, &normals)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn bandwidths(&self) -> Vec<f64> {
self.inner.bandwidths().to_vec()
}
fn n_samples(&self) -> usize {
self.inner.n_samples()
}
}
#[pyfunction]
#[pyo3(signature = (ssw, path, tracks=None))]
fn write_ssw(
ssw: &PySurfSrc,
path: &str,
tracks: Option<Vec<BTreeMap<String, f64>>>,
) -> PyResult<()> {
let header = ssw.inner.header.clone();
let track_data: Vec<nucleide_mcnp_io::surfsrc::TrackData> = match tracks {
Some(dict_tracks) => dict_tracks
.iter()
.map(|d| {
let g = |k: &str| d.get(k).copied().unwrap_or(0.0);
let mut record = vec![0.0f64; nucleide_mcnp_io::surfsrc::TrackData::RECORD_WIDTH];
record[0] = g("nps");
record[1] = g("bitarray");
record[2] = g("wgt");
record[3] = g("erg");
record[4] = g("tme");
record[5] = g("x");
record[6] = g("y");
record[7] = g("z");
record[8] = g("u");
record[9] = g("v");
record[10] = g("cs");
nucleide_mcnp_io::surfsrc::TrackData::from_record(record)
})
.collect(),
None => ssw
.inner
.read_tracklist()
.map_err(|e| PyValueError::new_err(e.to_string()))?,
};
let mut f = std::fs::File::create(path).map_err(|e| PyValueError::new_err(e.to_string()))?;
nucleide_mcnp_io::surfsrc::write_to(&mut f, &header, &track_data)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn mesh_to_geom(
x_bounds: Vec<f64>,
y_bounds: Vec<f64>,
z_bounds: Vec<f64>,
cell_materials: Vec<Option<(String, f64)>>,
title_card: &str,
) -> String {
let opts = nucleide_mcnp_io::deck::DeckOptions {
title_card: title_card.to_string(),
frac_type: nucleide_mcnp_io::deck::FracType::Mass,
};
nucleide_mcnp_io::deck::mesh_to_geom(&x_bounds, &y_bounds, &z_bounds, &cell_materials, &opts)
}
#[pyfunction]
fn alara_parse_deck(py: Python<'_>, text: &str) -> PyResult<Py<PyAny>> {
let owned = text.to_owned();
let deck = py
.detach(move || nucleide_alara_io::AlaraDeck::parse(&owned))
.map_err(ala_err)?;
Ok(deck_to_py(py, &deck))
}
fn ala_err(e: nucleide_alara_io::Error) -> PyErr {
PyValueError::new_err(e.to_string())
}
fn deck_to_py(py: Python<'_>, deck: &nucleide_alara_io::AlaraDeck) -> Py<PyAny> {
use pyo3::types::PyDict;
let out = PyDict::new(py);
let block_kinds: Vec<&str> = deck.block_kinds();
out.set_item("block_kinds", block_kinds).ok();
out.set_item("geometry", deck.geometry.as_ref().map(|g| g.kind.clone()))
.ok();
let mixtures: Vec<Py<PyAny>> = deck.mixtures.iter().map(|m| mixture_to_py(py, m)).collect();
out.set_item("mixtures", mixtures).ok();
let fluxes: Vec<Py<PyAny>> = deck.fluxes.iter().map(|f| fluxdef_to_py(py, f)).collect();
out.set_item("fluxes", fluxes).ok();
out.set_item(
"cooling_times_s",
deck.cooling
.as_ref()
.map(|c| c.times_s.clone())
.unwrap_or_default(),
)
.ok();
let schedules: Vec<Py<PyAny>> = deck
.schedules
.iter()
.map(|s| {
let d = PyDict::new(py);
let items: Vec<Vec<String>> = s.items.iter().map(|it| it.tokens.clone()).collect();
d.set_item("name", &s.name).ok();
d.set_item("items", items).ok();
d.into_any().unbind()
})
.collect();
out.set_item("schedules", schedules).ok();
let histories: Vec<Py<PyAny>> = deck
.pulse_histories
.iter()
.map(|h| {
let d = PyDict::new(py);
let levels: Vec<Py<PyAny>> = h
.levels
.iter()
.map(|l| {
let e = PyDict::new(py);
e.set_item("pulses", l.pulses).ok();
e.set_item("delay_s", l.delay_s).ok();
e.into_any().unbind()
})
.collect();
d.set_item("name", &h.name).ok();
d.set_item("levels", levels).ok();
d.into_any().unbind()
})
.collect();
out.set_item("pulse_histories", histories).ok();
let outputs: Vec<Py<PyAny>> = deck
.outputs
.iter()
.map(|o| {
let d = PyDict::new(py);
d.set_item("resolution", &o.resolution).ok();
d.set_item("entries", o.entries.clone()).ok();
d.into_any().unbind()
})
.collect();
out.set_item("outputs", outputs).ok();
out.set_item("truncation", deck.truncation.as_ref().map(|t| t.tolerance))
.ok();
out.into_any().unbind()
}
fn mixture_to_py(py: Python<'_>, mix: &nucleide_alara_io::deck::Mixture) -> Py<PyAny> {
use pyo3::types::PyDict;
let entries: Vec<Py<PyAny>> = mix
.entries
.iter()
.map(|e| mixture_entry_to_py(py, e))
.collect();
let d = PyDict::new(py);
d.set_item("name", &mix.name).ok();
d.set_item("entries", entries).ok();
d.into_any().unbind()
}
fn mixture_entry_to_py(py: Python<'_>, entry: &nucleide_alara_io::deck::MixtureEntry) -> Py<PyAny> {
use nucleide_alara_io::deck::MixtureEntry as E;
use pyo3::types::PyDict;
let d = PyDict::new(py);
match entry {
E::Material {
name,
rel_density,
vol_fraction,
} => {
d.set_item("kind", "material").ok();
d.set_item("name", name).ok();
d.set_item("rel_density", *rel_density).ok();
d.set_item("vol_fraction", *vol_fraction).ok();
}
E::Element {
symbol,
rel_density,
vol_fraction,
} => {
d.set_item("kind", "element").ok();
d.set_item("symbol", symbol).ok();
d.set_item("rel_density", *rel_density).ok();
d.set_item("vol_fraction", *vol_fraction).ok();
}
E::Like {
mixture,
rel_density,
} => {
d.set_item("kind", "like").ok();
d.set_item("mixture", mixture).ok();
d.set_item("rel_density", *rel_density).ok();
}
E::Target { target_kind, name } => {
d.set_item("kind", "target").ok();
d.set_item("target_kind", target_kind).ok();
d.set_item("name", name).ok();
}
}
d.into_any().unbind()
}
fn fluxdef_to_py(py: Python<'_>, flux: &nucleide_alara_io::deck::FluxDef) -> Py<PyAny> {
use pyo3::types::PyDict;
let d = PyDict::new(py);
d.set_item("name", &flux.name).ok();
d.set_item("file", &flux.file).ok();
d.set_item("scale", flux.scale).ok();
d.set_item("skip", flux.skip).ok();
d.set_item("format", &flux.format).ok();
d.into_any().unbind()
}
#[pyfunction]
fn alara_parse_flux(py: Python<'_>, text: &str, name: &str) -> PyResult<Py<PyAny>> {
let owned_text = text.to_owned();
let owned_name = name.to_owned();
let spectra = py
.detach(move || nucleide_alara_io::FluxSpectra::parse(&owned_name, &owned_text))
.map_err(ala_err)?;
use pyo3::types::PyDict;
let d = PyDict::new(py);
d.set_item("name", spectra.name.clone()).ok();
d.set_item("groups_per_interval", spectra.groups_per_interval)
.ok();
d.set_item("num_intervals", spectra.num_intervals()).ok();
let totals: Vec<f64> = spectra.intervals.iter().map(|iv| iv.iter().sum()).collect();
d.set_item("totals", totals).ok();
d.set_item("total", spectra.total()).ok();
d.set_item("intervals", spectra.intervals.clone()).ok();
Ok(d.into_any().unbind())
}
#[pyfunction]
fn alara_parse_output(
py: Python<'_>,
text: &str,
run_lbl: &str,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let owned_text = text.to_owned();
let owned_lbl = run_lbl.to_owned();
let rows = py
.detach(move || {
nucleide_alara_io::output::ResponseFrame::parse(&owned_text, &owned_lbl).map(|f| f.rows)
})
.map_err(ala_err)?;
Ok(rows
.iter()
.map(|r| {
let mut d = BTreeMap::new();
d.insert(
"time_s".to_string(),
r.time_s.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"time_label".to_string(),
r.time_label
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"nuclide".to_string(),
r.nuclide
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"half_life_s".to_string(),
r.half_life_s.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"run_lbl".to_string(),
r.run_lbl
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"block".to_string(),
r.block
.as_str()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"block_name".to_string(),
r.block_name
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"block_num".to_string(),
r.block_num.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"variable".to_string(),
r.variable
.as_str()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"var_unit".to_string(),
r.var_unit
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"value".to_string(),
r.value.into_pyobject(py).unwrap().unbind().into_any(),
);
d
})
.collect())
}
#[pyfunction]
#[pyo3(signature = (deck_text, top=None))]
fn alara_expand_schedule(
py: Python<'_>,
deck_text: &str,
top: Option<&str>,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let owned_text = deck_text.to_owned();
let owned_top = top.map(str::to_owned);
let steps = py
.detach(move || expand_deck_schedules(&owned_text, owned_top.as_deref()))
.map_err(PyValueError::new_err)?;
Ok(steps
.into_iter()
.map(|s| {
let mut d = BTreeMap::new();
let cooling = s.is_cooling();
d.insert(
"duration_s".to_string(),
s.duration_s.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"flux".to_string(),
s.flux
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"is_cooling".to_string(),
pyo3::types::PyBool::new(py, cooling)
.to_owned()
.into_any()
.unbind(),
);
d
})
.collect())
}
fn expand_deck_schedules(
deck_text: &str,
top: Option<&str>,
) -> Result<Vec<nucleide_alara_io::FlatStep>, String> {
let deck = nucleide_alara_io::AlaraDeck::parse(deck_text).map_err(|e| e.to_string())?;
let mut scheds = Vec::with_capacity(deck.schedules.len());
for raw in &deck.schedules {
let mut items = Vec::with_capacity(raw.items.len());
for entry in &raw.items {
items.push(
parse_deck_sched_item(&entry.tokens)
.map_err(|m| format!("schedule `{}` line {}: {m}", raw.name, entry.line))?,
);
}
scheds.push(nucleide_alara_io::schedule::ScheduleDef {
name: raw.name.clone(),
items,
});
}
let histories: Vec<nucleide_alara_io::schedule::PulseHistory> = deck
.pulse_histories
.iter()
.map(|h| nucleide_alara_io::schedule::PulseHistory {
name: h.name.clone(),
levels: h
.levels
.iter()
.map(|l| nucleide_alara_io::schedule::PulseLevel {
count: l.pulses,
delay_s: l.delay_s,
})
.collect(),
})
.collect();
match top {
Some(name) => {
nucleide_alara_io::expand_from(name, &scheds, &histories).map_err(|e| e.to_string())
}
None => nucleide_alara_io::expand(&scheds, &histories).map_err(|e| e.to_string()),
}
}
fn parse_deck_sched_item(tokens: &[String]) -> Result<nucleide_alara_io::SchedItem, String> {
match tokens {
[op_text, op_unit, flux, history, delay_text, delay_unit] => {
let op: f64 = op_text
.parse()
.map_err(|_| format!("expected operating time, found `{op_text}`"))?;
let delay: f64 = delay_text
.parse()
.map_err(|_| format!("expected delay, found `{delay_text}`"))?;
let op_time_s =
nucleide_alara_io::parse_time_to_seconds(op, op_unit).map_err(|e| e.to_string())?;
let delay_s = nucleide_alara_io::parse_time_to_seconds(delay, delay_unit)
.map_err(|e| e.to_string())?;
Ok(nucleide_alara_io::SchedItem::Pulse {
op_time_s,
flux: flux.clone(),
history: history.clone(),
delay_s,
})
}
[name, history, delay_text, delay_unit] => {
let delay: f64 = delay_text
.parse()
.map_err(|_| format!("expected delay, found `{delay_text}`"))?;
let delay_s = nucleide_alara_io::parse_time_to_seconds(delay, delay_unit)
.map_err(|e| e.to_string())?;
Ok(nucleide_alara_io::SchedItem::SubSchedule {
name: name.clone(),
history: history.clone(),
delay_s,
})
}
_ => Err(format!(
"expected 4- or 6-token schedule item, found {}",
tokens.join(" ")
)),
}
}
#[pyfunction]
fn half_life(key: &Bound<'_, PyAny>) -> PyResult<Option<f64>> {
lookup(key, nucleide_nuclei::data::half_life)
}
#[pyfunction]
fn decay_constant(key: &Bound<'_, PyAny>) -> PyResult<Option<f64>> {
lookup(key, nucleide_nuclei::data::decay_constant)
}
#[pyfunction]
fn q_value_capture(key: &Bound<'_, PyAny>) -> PyResult<Option<f64>> {
lookup(key, nucleide_nuclei::data::q_value_neutron_capture)
}
#[pyfunction]
fn q_value_alpha(key: &Bound<'_, PyAny>) -> PyResult<Option<f64>> {
lookup(key, nucleide_nuclei::data::q_value_alpha)
}
#[pyfunction]
fn read_inp(path: &str) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let mats = nucleide_mcnp_io::inp::materials_from_file(path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Python::attach(|py| {
Ok(mats
.into_iter()
.map(|m| {
let mut d = BTreeMap::new();
d.insert(
"number".to_string(),
m.number.into_pyobject(py).unwrap().unbind().into_any(),
);
let fr: BTreeMap<String, f64> = m
.fractions
.iter()
.map(|(id, f)| (id.to_name(), *f))
.collect();
d.insert(
"fractions".to_string(),
fr.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"fraction_type".to_string(),
match m.fraction_type {
nucleide_mcnp_io::inp::FracKind::Atom => "atom",
nucleide_mcnp_io::inp::FracKind::Mass => "mass",
}
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"density".to_string(),
m.density.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"comments".to_string(),
m.comments
.join(" ")
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d
})
.collect())
})
}
fn comp_to_material(comp: BTreeMap<String, f64>) -> PyResult<nucleide_material::Material> {
let mut mat = nucleide_material::Material::new();
for (name, grams) in &comp {
let id = nucleide_nuclei::NuclideId::from_name(name)
.map_err(|e| PyValueError::new_err(format!("`{name}`: {e}")))?;
mat.add_nuclide(id, *grams);
}
Ok(mat)
}
#[pyfunction]
fn from_formula(formula: &str) -> PyResult<BTreeMap<String, f64>> {
use nucleide_material::AbundanceProvider;
let parsed = nucleide_material::parse_formula(formula)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let mut nat = Vec::new();
for (z, count) in &parsed {
if let Some(isotopes) = nucleide_material::NaturalAbundances.natural_isotopes(*z) {
for (id, frac) in isotopes {
nat.push((id, frac * count));
}
}
}
let total: f64 = nat.iter().map(|(_, c)| c).sum();
if total <= 0.0 {
return Err(PyValueError::new_err("empty formula expansion"));
}
let mut out: BTreeMap<String, f64> = BTreeMap::new();
for (id, atoms) in nat {
*out.entry(id.to_name()).or_insert(0.0) += atoms / total;
}
Ok(out)
}
#[pyfunction]
fn activity(comp: BTreeMap<String, f64>) -> PyResult<BTreeMap<String, f64>> {
let mat = comp_to_material(comp)?;
let analytics = nucleide_material::Analytics {
masses: &nucleide_material::Ame2020,
decays: &nucleide_material::ChainDecays,
};
let per_nuc = mat
.activity(&analytics)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let specific = mat
.specific_activity(&analytics)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let mut out: BTreeMap<String, f64> = per_nuc
.into_iter()
.map(|(id, v)| (id.to_name(), v))
.collect();
out.insert("specific".to_string(), specific);
Ok(out)
}
#[pyfunction]
fn to_xml(comp: BTreeMap<String, f64>, name: &str, density: f64, units: &str) -> PyResult<String> {
let mat = comp_to_material(comp)?;
mat.to_xml(name, density, units)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyclass(name = "Cascade")]
struct PyCascade {
inner: std::sync::Mutex<nucleide_enrichment::Cascade>,
}
#[pymethods]
impl PyCascade {
#[staticmethod]
fn default_uranium() -> Self {
Self {
inner: std::sync::Mutex::new(nucleide_enrichment::default_uranium_cascade()),
}
}
#[new]
#[allow(non_snake_case)]
#[allow(clippy::too_many_arguments)]
fn new(
alpha: f64,
Mstar: f64,
j: u32,
k: u32,
N: f64,
M: f64,
x_feed_j: f64,
x_prod_j: f64,
x_tail_j: f64,
mat_feed: BTreeMap<String, f64>,
) -> PyResult<Self> {
let mut feed = BTreeMap::new();
for (name, frac) in mat_feed {
let id = NuclideId::from_name(&name).map_err(wrap_nucid_err)?;
feed.insert(id, frac);
}
let casc = nucleide_enrichment::Cascade {
alpha,
Mstar,
j: NuclideId::from_nucid(j),
k: NuclideId::from_nucid(k),
N,
M,
x_feed_j,
x_prod_j,
x_tail_j,
mat_feed: nucleide_enrichment::Stream::with_total_mass(feed, 1.0),
mat_prod: nucleide_enrichment::Stream::new(),
mat_tail: nucleide_enrichment::Stream::new(),
l_t_per_feed: 0.0,
swu_per_feed: 0.0,
swu_per_prod: 0.0,
};
Ok(Self {
inner: std::sync::Mutex::new(casc),
})
}
#[pyo3(signature = (tolerance=None, max_iterations=None))]
fn solve(&self, tolerance: Option<f64>, max_iterations: Option<u32>) -> PyResult<()> {
let tol = tolerance.unwrap_or(nucleide_enrichment::DEFAULT_TOLERANCE);
let iters = max_iterations.unwrap_or(nucleide_enrichment::DEFAULT_MAX_ITER);
let mut c = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?;
*c = nucleide_enrichment::solve_numeric(&c, tol, iters)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(())
}
#[pyo3(signature = (tolerance=None, max_iterations=None))]
fn solve_multicomponent(
&self,
tolerance: Option<f64>,
max_iterations: Option<u32>,
) -> PyResult<()> {
let tol = tolerance.unwrap_or(nucleide_enrichment::DEFAULT_TOLERANCE);
let iters = max_iterations.unwrap_or(nucleide_enrichment::DEFAULT_MAX_ITER);
let mut c = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?;
*c = nucleide_enrichment::multicomponent(&c, tol, iters)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(())
}
#[getter]
fn alpha(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.alpha)
}
#[getter]
#[allow(non_snake_case)]
fn Mstar(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.Mstar)
}
#[getter]
#[allow(non_snake_case)]
fn N(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.N)
}
#[getter]
#[allow(non_snake_case)]
fn M(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.M)
}
#[getter]
fn x_feed_j(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.x_feed_j)
}
#[getter]
fn x_prod_j(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.x_prod_j)
}
#[getter]
fn x_tail_j(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.x_tail_j)
}
#[getter]
fn l_t_per_feed(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.l_t_per_feed)
}
#[getter]
fn swu_per_feed(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.swu_per_feed)
}
#[getter]
fn swu_per_prod(&self) -> PyResult<f64> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.swu_per_prod)
}
#[getter]
fn mat_feed(&self) -> PyResult<BTreeMap<String, f64>> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.mat_feed
.comp
.iter()
.map(|(id, frac)| (id.to_name(), *frac))
.collect())
}
#[getter]
fn mat_prod(&self) -> PyResult<BTreeMap<String, f64>> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.mat_prod
.comp
.iter()
.map(|(id, frac)| (id.to_name(), *frac))
.collect())
}
#[getter]
fn mat_tail(&self) -> PyResult<BTreeMap<String, f64>> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?
.mat_tail
.comp
.iter()
.map(|(id, frac)| (id.to_name(), *frac))
.collect())
}
fn separative_work_per_product(&self) -> PyResult<f64> {
let c = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?;
Ok(nucleide_enrichment::swu_per_prod(
c.x_feed_j, c.x_prod_j, c.x_tail_j,
))
}
fn __repr__(&self) -> PyResult<String> {
let c = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("cascade lock poisoned"))?;
Ok(format!(
"Cascade(alpha={}, Mstar={}, x_prod_j={:.5})",
c.alpha, c.Mstar, c.x_prod_j
))
}
}
#[pyfunction]
fn enrichment_value_func(x: f64) -> f64 {
nucleide_enrichment::value_func(x)
}
#[pyfunction]
fn enrichment_swu_per_feed(x_feed: f64, x_prod: f64, x_tail: f64) -> f64 {
nucleide_enrichment::swu_per_feed(x_feed, x_prod, x_tail)
}
#[pyfunction]
fn enrichment_swu_per_prod(x_feed: f64, x_prod: f64, x_tail: f64) -> f64 {
nucleide_enrichment::swu_per_prod(x_feed, x_prod, x_tail)
}
#[pyfunction]
fn enrichment_swu_per_tail(x_feed: f64, x_prod: f64, x_tail: f64) -> f64 {
nucleide_enrichment::swu_per_tail(x_feed, x_prod, x_tail)
}
#[pyclass(name = "MaterialsCompendium")]
struct PyMaterialsCompendium {
inner: nucleide_material::MaterialsLibrary,
}
#[pymethods]
impl PyMaterialsCompendium {
#[staticmethod]
fn load(path: &str) -> PyResult<Self> {
nucleide_material::MaterialsLibrary::from_file(path)
.map(|inner| PyMaterialsCompendium { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn __len__(&self) -> usize {
self.inner.len()
}
fn names(&self) -> Vec<String> {
self.inner.names().into_iter().map(String::from).collect()
}
#[pyo3(signature = (name, as_material=false))]
#[allow(clippy::type_complexity)]
fn get(&self, name: &str, as_material: bool) -> PyResult<Option<BTreeMap<String, Py<PyAny>>>> {
let entry = match self.inner.get(name) {
Some(e) => e,
None => return Ok(None),
};
let named_fractions = if as_material {
Some(
entry
.to_material()
.map_err(|e| PyValueError::new_err(e.to_string()))?,
)
} else {
None
};
Ok(Python::attach(|py| {
let mut d: BTreeMap<String, Py<PyAny>> = BTreeMap::new();
d.insert(
"name".into(),
entry
.name
.as_str()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"mat_num".into(),
entry.mat_num.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"density".into(),
entry.density.into_pyobject(py).unwrap().unbind().into_any(),
);
match &named_fractions {
Some(mat) => {
let fr: BTreeMap<String, f64> =
mat.comp.iter().map(|(id, g)| (id.to_name(), *g)).collect();
d.insert(
"fractions".into(),
fr.into_pyobject(py).unwrap().unbind().into_any(),
);
}
None => {
let fr = entry.weight_fractions();
d.insert(
"fractions".into(),
fr.into_pyobject(py).unwrap().unbind().into_any(),
);
}
}
Some(d)
}))
}
}
#[pyfunction]
fn isotxs_parse(py: Python<'_>, text: &str) -> PyResult<Py<PyAny>> {
let owned = text.to_owned();
let lib = py
.detach(move || nucleide_cccc_io::IsotxsLib::parse(&owned))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(isotxs_to_py(py, &lib))
}
fn isotxs_to_py(py: Python<'_>, lib: &nucleide_cccc_io::IsotxsLib) -> Py<PyAny> {
use pyo3::types::PyDict;
let out = PyDict::new(py);
let nuclides: Vec<Py<PyAny>> = lib
.nuclides
.iter()
.map(|n| {
let d = PyDict::new(py);
d.set_item("label", &n.label).ok();
d.set_item("zaid", &n.zaid).ok();
d.set_item("groups", n.groups).ok();
d.set_item("total_xs", n.total_xs.clone()).ok();
d.into_any().unbind()
})
.collect();
out.set_item("nuclides", nuclides).ok();
out.into_any().unbind()
}
#[pyfunction]
#[pyo3(signature = (text, kind="rtflux"))]
fn rtflux_parse(py: Python<'_>, text: &str, kind: &str) -> PyResult<Py<PyAny>> {
let flux_kind = match kind.to_ascii_lowercase().as_str() {
"rtflux" => nucleide_cccc_io::rtflux::FluxKind::Rtflux,
"atflux" => nucleide_cccc_io::rtflux::FluxKind::Atflux,
"rzflux" => nucleide_cccc_io::rtflux::FluxKind::Rzflux,
other => {
return Err(PyValueError::new_err(format!(
"kind must be rtflux|atflux|rzflux, got `{other}`"
)))
}
};
let owned = text.to_owned();
let flux = py
.detach(move || nucleide_cccc_io::FluxFile::parse(flux_kind, &owned))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let d = PyDict::new(py);
d.set_item("kind", flux.kind.keyword()).ok();
d.set_item("groups", flux.groups).ok();
d.set_item("per_point", flux.per_point).ok();
d.set_item("npoints", flux.npoints()).ok();
d.set_item("values", flux.values.clone()).ok();
d.set_item("total", flux.total()).ok();
Ok(d.into_any().unbind())
}
fn partisn_deck_from_dict(
deck: &Bound<'_, pyo3::types::PyDict>,
) -> PyResult<nucleide_cccc_io::PartisnDeck> {
let title: String = match deck.get_item("title")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("partisn deck `title` must be str"))?,
None => return Err(PyValueError::new_err("partisn deck missing `title`")),
};
let dim: u8 = match deck.get_item("dim")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("partisn deck `dim` must be 1, 2, or 3"))?,
None => return Err(PyValueError::new_err("partisn deck missing `dim`")),
};
let zones_value = match deck.get_item("zones")? {
Some(v) => v,
None => return Err(PyValueError::new_err("partisn deck missing `zones`")),
};
let zone_dicts: Vec<Bound<'_, pyo3::types::PyDict>> = zones_value
.extract()
.map_err(|_| PyValueError::new_err("partisn deck `zones` must be a list of dicts"))?;
let mut zones = Vec::with_capacity(zone_dicts.len());
for z in &zone_dicts {
let id: u32 = match z.get_item("id")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("partisn zone `id` must be int"))?,
None => return Err(PyValueError::new_err("partisn zone missing `id`")),
};
let material: String = match z.get_item("material")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("partisn zone `material` must be str"))?,
None => return Err(PyValueError::new_err("partisn zone missing `material`")),
};
let isotxs_labels: Vec<String> = match z.get_item("isotxs_labels")? {
Some(v) => v.extract().map_err(|_| {
PyValueError::new_err("partisn zone `isotxs_labels` must be a list of str")
})?,
None => {
return Err(PyValueError::new_err(
"partisn zone missing `isotxs_labels`",
))
}
};
let density: f64 = match z.get_item("density")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("partisn zone `density` must be float"))?,
None => return Err(PyValueError::new_err("partisn zone missing `density`")),
};
zones.push(nucleide_cccc_io::partisn::PartisnZone {
id,
material,
isotxs_labels,
density,
});
}
let source: Option<String> = match deck.get_item("source")? {
Some(v) if v.is_none() => None,
Some(v) => Some(
v.extract()
.map_err(|_| PyValueError::new_err("partisn deck `source` must be str or None"))?,
),
None => None,
};
Ok(nucleide_cccc_io::PartisnDeck {
title,
dim,
zones,
source,
})
}
#[pyfunction]
fn partisn_render(py: Python<'_>, deck: &Bound<'_, pyo3::types::PyDict>) -> PyResult<String> {
let rust_deck = partisn_deck_from_dict(deck)?;
Ok(py.detach(move || rust_deck.render()))
}
#[pyfunction]
fn partisn_validate(
py: Python<'_>,
deck: &Bound<'_, pyo3::types::PyDict>,
isotxs_text: &str,
) -> PyResult<()> {
let rust_deck = partisn_deck_from_dict(deck)?;
let owned = isotxs_text.to_owned();
let lib = py
.detach(move || nucleide_cccc_io::IsotxsLib::parse(&owned))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
rust_deck
.validate(&lib)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn fispact_row_to_map(
py: Python<'_>,
r: &nucleide_alara_io::output::ResponseRow,
) -> BTreeMap<String, Py<PyAny>> {
let mut d = BTreeMap::new();
d.insert(
"time_s".to_string(),
r.time_s.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"time_label".to_string(),
r.time_label
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"nuclide".to_string(),
r.nuclide
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"half_life_s".to_string(),
r.half_life_s.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"run_lbl".to_string(),
r.run_lbl
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"block".to_string(),
r.block
.as_str()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"block_name".to_string(),
r.block_name
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"block_num".to_string(),
r.block_num.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"variable".to_string(),
r.variable
.as_str()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"var_unit".to_string(),
r.var_unit
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"value".to_string(),
r.value.into_pyobject(py).unwrap().unbind().into_any(),
);
d
}
#[pyfunction]
fn fispact_parse_output(
py: Python<'_>,
text: &str,
run_lbl: &str,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let owned_text = text.to_owned();
let owned_lbl = run_lbl.to_owned();
let rows = py
.detach(move || {
nucleide_fispact_io::parse_to_frame(&owned_text, &owned_lbl).map(|f| f.rows)
})
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(rows.iter().map(|r| fispact_row_to_map(py, r)).collect())
}
#[pyfunction]
fn fispact_parse_clearance(
py: Python<'_>,
text: &str,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let owned_text = text.to_owned();
let scan = py
.detach(move || nucleide_fispact_io::parse_clearance(&owned_text))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(scan
.rows
.iter()
.map(|row| {
let mut d = BTreeMap::new();
d.insert(
"interval".to_string(),
row.interval.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"time_s".to_string(),
row.time_s.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"time_label".to_string(),
row.time_label
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"cooling".to_string(),
pyo3::types::PyBool::new(py, row.cooling)
.to_owned()
.into_any()
.unbind(),
);
d.insert(
"nuclide".to_string(),
row.nuclide
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"flags".to_string(),
row.flags
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"activity_bq".to_string(),
row.activity_bq
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"clearance_index".to_string(),
row.clearance_index
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"half_life_s".to_string(),
row.half_life_s
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d
})
.collect())
}
#[pyfunction]
fn origen_parse_tape5(py: Python<'_>, text: &str) -> PyResult<Py<PyAny>> {
let owned = text.to_owned();
let tape = py
.detach(move || nucleide_origen_io::Tape5::parse(&owned))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("titles", tape.titles.clone()).ok();
let steps: Vec<Py<PyAny>> = tape
.irradiation_steps
.iter()
.map(|s| {
let d = PyDict::new(py);
d.set_item("flux", s.flux).ok();
d.set_item("days", s.days).ok();
d.into_any().unbind()
})
.collect();
out.set_item("irradiation_steps", steps).ok();
let materials: Vec<Py<PyAny>> = tape
.materials
.iter()
.map(|m| {
let d = PyDict::new(py);
d.set_item("name", &m.name).ok();
let entries: Vec<Py<PyAny>> = m
.grams
.iter()
.map(|(nuclide, grams)| {
let e = PyDict::new(py);
e.set_item("nuclide", nuclide).ok();
e.set_item("grams", *grams).ok();
e.into_any().unbind()
})
.collect();
d.set_item("entries", entries).ok();
d.into_any().unbind()
})
.collect();
out.set_item("materials", materials).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
fn origen_parse_tape6(py: Python<'_>, text: &str) -> PyResult<Py<PyAny>> {
let owned = text.to_owned();
let tape = py
.detach(move || nucleide_origen_io::Tape6::parse(&owned))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
let records: Vec<Py<PyAny>> = tape
.records
.iter()
.map(|r| {
let d = PyDict::new(py);
d.set_item("nuclide", &r.nuclide).ok();
d.set_item("grams", r.grams).ok();
d.set_item("activity_bq", r.activity_bq).ok();
d.into_any().unbind()
})
.collect();
out.set_item("records", records).ok();
out.set_item("total_activity", tape.total_activity()).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
fn origen_parse_tape9(py: Python<'_>, text: &str) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let owned = text.to_owned();
let entries = py
.detach(move || nucleide_origen_io::Tape9Entry::parse(&owned))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(entries
.iter()
.map(|e| {
let mut d = BTreeMap::new();
d.insert(
"nuclide".to_string(),
e.nuclide
.clone()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"decay_const".to_string(),
e.decay_const.into_pyobject(py).unwrap().unbind().into_any(),
);
d
})
.collect())
}
fn r2s_workflow_to_py(py: Python<'_>, workflow: &nucleide_r2s::R2sWorkflow) -> Py<PyAny> {
use pyo3::types::PyDict;
let out = PyDict::new(py);
let steps: Vec<Py<PyAny>> = workflow
.steps
.iter()
.map(|s| {
let d = PyDict::new(py);
d.set_item("zone", &s.zone).ok();
d.set_item("flux", &s.flux).ok();
d.into_any().unbind()
})
.collect();
out.set_item("steps", steps).ok();
out.set_item("cooling_s", workflow.cooling_s.clone()).ok();
out.set_item("top_schedule", &workflow.top_schedule).ok();
out.into_any().unbind()
}
fn r2s_workflow_from_dict(
workflow: &Bound<'_, pyo3::types::PyDict>,
) -> PyResult<nucleide_r2s::R2sWorkflow> {
let steps_value = match workflow.get_item("steps")? {
Some(v) => v,
None => return Err(PyValueError::new_err("r2s workflow missing `steps`")),
};
let step_dicts: Vec<Bound<'_, pyo3::types::PyDict>> = steps_value
.extract()
.map_err(|_| PyValueError::new_err("r2s workflow `steps` must be a list of dicts"))?;
let mut steps = Vec::with_capacity(step_dicts.len());
for s in &step_dicts {
let zone: String = match s.get_item("zone")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("r2s step `zone` must be str"))?,
None => return Err(PyValueError::new_err("r2s step missing `zone`")),
};
let flux: String = match s.get_item("flux")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("r2s step `flux` must be str"))?,
None => return Err(PyValueError::new_err("r2s step missing `flux`")),
};
steps.push(nucleide_r2s::R2sStep { zone, flux });
}
let cooling_s: Vec<f64> = match workflow.get_item("cooling_s")? {
Some(v) => v.extract().map_err(|_| {
PyValueError::new_err("r2s workflow `cooling_s` must be a list of float")
})?,
None => return Err(PyValueError::new_err("r2s workflow missing `cooling_s`")),
};
let top_schedule: String = match workflow.get_item("top_schedule")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("r2s workflow `top_schedule` must be str"))?,
None => return Err(PyValueError::new_err("r2s workflow missing `top_schedule`")),
};
Ok(nucleide_r2s::R2sWorkflow {
steps,
cooling_s,
top_schedule,
})
}
#[pyfunction]
fn r2s_from_deck(py: Python<'_>, deck_text: &str) -> PyResult<Py<PyAny>> {
let owned = deck_text.to_owned();
let workflow = py
.detach(move || {
let deck = nucleide_alara_io::AlaraDeck::parse(&owned)
.map_err(|e| nucleide_r2s::Error::Invalid(e.to_string()))?;
nucleide_r2s::R2sWorkflow::from_deck(&deck)
})
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(r2s_workflow_to_py(py, &workflow))
}
#[pyfunction]
fn r2s_validate(
py: Python<'_>,
workflow: &Bound<'_, pyo3::types::PyDict>,
deck_text: &str,
) -> PyResult<()> {
let rust_workflow = r2s_workflow_from_dict(workflow)?;
let owned = deck_text.to_owned();
let deck = py
.detach(move || nucleide_alara_io::AlaraDeck::parse(&owned))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
rust_workflow
.validate_against(&deck)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (deck_text, top=None))]
fn r2s_expand(
py: Python<'_>,
deck_text: &str,
top: Option<&str>,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let owned_text = deck_text.to_owned();
let owned_top = top.map(str::to_owned);
let steps = py
.detach(move || {
let deck = nucleide_alara_io::AlaraDeck::parse(&owned_text)
.map_err(|e| nucleide_r2s::Error::Invalid(e.to_string()))?;
let mut workflow = nucleide_r2s::R2sWorkflow::from_deck(&deck)?;
if let Some(top) = owned_top {
workflow.top_schedule = top;
}
workflow.expand(&deck, &[])
})
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(steps
.into_iter()
.map(|s| {
let mut d = BTreeMap::new();
let cooling = s.is_cooling();
d.insert(
"duration_s".to_string(),
s.duration_s.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"flux".to_string(),
s.flux.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"is_cooling".to_string(),
pyo3::types::PyBool::new(py, cooling)
.to_owned()
.into_any()
.unbind(),
);
d
})
.collect())
}
#[pyfunction]
fn r2s_assemble(
py: Python<'_>,
output_text: &str,
run_lbl: &str,
zone: &str,
groups: usize,
) -> PyResult<Py<PyAny>> {
let owned_text = output_text.to_owned();
let owned_lbl = run_lbl.to_owned();
let owned_zone = zone.to_owned();
let source = py
.detach(move || {
let frame = nucleide_alara_io::output::ResponseFrame::parse(&owned_text, &owned_lbl)
.map_err(|e| nucleide_r2s::Error::Invalid(e.to_string()))?;
Ok::<_, nucleide_r2s::Error>(nucleide_r2s::photon::assemble(
&frame,
&owned_zone,
groups,
))
})
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("zone", source.zone.clone()).ok();
out.set_item("groups", source.groups.clone()).ok();
out.set_item("total", source.total()).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
#[pyo3(signature = (totals, zone_of_voxel, split=false))]
fn r2s_tag_zone_strength(
py: Python<'_>,
totals: Vec<f64>,
zone_of_voxel: Vec<usize>,
split: bool,
) -> PyResult<Py<PyAny>> {
let zones: Vec<nucleide_r2s::photon::ZonePhotonSource> = totals
.into_iter()
.enumerate()
.map(|(i, total)| {
let groups = if total == 0.0 {
Vec::new()
} else {
vec![total]
};
nucleide_r2s::photon::ZonePhotonSource {
zone: format!("zone{i}"),
groups,
}
})
.collect();
let tags = if split {
nucleide_r2s::tags::split_zone_totals(&zones, &zone_of_voxel)
} else {
nucleide_r2s::tags::tag_zone_totals(&zones, &zone_of_voxel)
}
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("n_zones", tags.n_zones).ok();
out.set_item("zone_of_voxel", tags.zone_of_voxel.clone())
.ok();
out.set_item("source_strength", tags.source_strength.clone())
.ok();
out.set_item("decay_time_s", tags.decay_time_s.clone()).ok();
out.set_item("total", tags.total_strength()).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
fn r2s_photon_group_sums(
py: Python<'_>,
photon_text: &str,
nuclides: Vec<String>,
time_s: f64,
) -> PyResult<Py<PyAny>> {
let source = nucleide_alara_io::photon::PhotonSource::from_str(photon_text)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let names: Vec<&str> = nuclides.iter().map(String::as_str).collect();
let at = nucleide_r2s::tags::photon_groups_at(&source, &names, time_s);
let sums = nucleide_r2s::tags::sum_group_strengths(&at)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
let rows: Vec<Py<PyAny>> = at
.iter()
.map(|g| {
let d = PyDict::new(py);
d.set_item("nuclide", g.nuclide.clone()).ok();
d.set_item("time_s", g.time_s).ok();
d.set_item("strengths", g.strengths.clone()).ok();
d.into_any().unbind()
})
.collect();
out.set_item("groups", rows).ok();
out.set_item("sums", sums.clone()).ok();
out.set_item("total", sums.iter().sum::<f64>()).ok();
Ok(out.into_any().unbind())
}
fn snapshot_dict_str(
zone: &Bound<'_, pyo3::types::PyDict>,
key: &str,
what: &str,
) -> PyResult<String> {
match zone.get_item(key)? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err(format!("snapshot {what} `{key}` must be str"))),
None => Err(PyValueError::new_err(format!(
"snapshot {what} missing `{key}`"
))),
}
}
fn snapshot_dict_opt_str(
zone: &Bound<'_, pyo3::types::PyDict>,
key: &str,
what: &str,
) -> PyResult<Option<String>> {
match zone.get_item(key)? {
Some(v) if v.is_none() => Ok(None),
Some(v) => v
.extract::<String>()
.map(Some)
.map_err(|_| PyValueError::new_err(format!("snapshot {what} `{key}` must be str"))),
None => Ok(None),
}
}
fn snapshot_dict_f64(
zone: &Bound<'_, pyo3::types::PyDict>,
key: &str,
what: &str,
) -> PyResult<f64> {
match zone.get_item(key)? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err(format!("snapshot {what} `{key}` must be float"))),
None => Err(PyValueError::new_err(format!(
"snapshot {what} missing `{key}`"
))),
}
}
fn snapshot_dict_opt_f64(
zone: &Bound<'_, pyo3::types::PyDict>,
key: &str,
what: &str,
) -> PyResult<Option<f64>> {
match zone.get_item(key)? {
Some(v) if v.is_none() => Ok(None),
Some(v) => v
.extract::<f64>()
.map(Some)
.map_err(|_| PyValueError::new_err(format!("snapshot {what} `{key}` must be float"))),
None => Ok(None),
}
}
fn snapshot_zone_from_dict(
zone: &Bound<'_, pyo3::types::PyDict>,
) -> PyResult<nucleide_r2s::snapshot::SnapshotZone> {
let id = snapshot_dict_str(zone, "id", "zone")?;
let volume_cm3 = snapshot_dict_f64(zone, "volume_cm3", "zone")?;
let composition: BTreeMap<String, f64> = match zone.get_item("composition")? {
Some(v) => v.extract().map_err(|_| {
PyValueError::new_err("snapshot zone `composition` must be a dict of str to float")
})?,
None => return Err(PyValueError::new_err("snapshot zone missing `composition`")),
};
Ok(nucleide_r2s::snapshot::SnapshotZone {
zone: id,
volume_cm3,
zbottom_cm: snapshot_dict_opt_f64(zone, "zbottom_cm", "zone")?,
ztop_cm: snapshot_dict_opt_f64(zone, "ztop_cm", "zone")?,
material: snapshot_dict_opt_str(zone, "material", "zone")?,
xs_type: snapshot_dict_opt_str(zone, "xs_type", "zone")?,
temperature_c: snapshot_dict_opt_f64(zone, "temperature_C", "zone")?,
composition: composition.into_iter().collect(),
flux_name: snapshot_dict_opt_str(zone, "flux", "zone")?,
})
}
fn snapshot_input_from_dict(
snapshot: &Bound<'_, pyo3::types::PyDict>,
) -> PyResult<nucleide_r2s::snapshot::SnapshotInput> {
let zone_dicts: Vec<Bound<'_, pyo3::types::PyDict>> = match snapshot.get_item("zones")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("snapshot `zones` must be a list of dicts"))?,
None => return Err(PyValueError::new_err("snapshot missing `zones`")),
};
let mut zones = Vec::with_capacity(zone_dicts.len());
for z in &zone_dicts {
zones.push(snapshot_zone_from_dict(z)?);
}
let flux_dicts: Vec<Bound<'_, pyo3::types::PyDict>> = match snapshot.get_item("flux_defs")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("snapshot `flux_defs` must be a list of dicts"))?,
None => return Err(PyValueError::new_err("snapshot missing `flux_defs`")),
};
let mut flux_defs = Vec::with_capacity(flux_dicts.len());
for f in &flux_dicts {
flux_defs.push(nucleide_r2s::snapshot::SnapshotFluxDef {
name: snapshot_dict_str(f, "name", "flux")?,
file: snapshot_dict_str(f, "file", "flux")?,
scale: snapshot_dict_f64(f, "scale", "flux")?,
});
}
let cooling_s: Vec<f64> = match snapshot.get_item("cooling_s")? {
Some(v) => v
.extract()
.map_err(|_| PyValueError::new_err("snapshot `cooling_s` must be a list of float"))?,
None => return Err(PyValueError::new_err("snapshot missing `cooling_s`")),
};
Ok(nucleide_r2s::snapshot::SnapshotInput {
zones,
flux_defs,
cooling_s,
schedule_text: snapshot_dict_opt_str(snapshot, "schedule_text", "snapshot")?,
output: snapshot_dict_opt_str(snapshot, "output", "snapshot")?,
})
}
#[pyfunction]
fn r2s_snapshot_inventory(
snapshot: &Bound<'_, pyo3::types::PyDict>,
) -> PyResult<BTreeMap<String, f64>> {
let input = snapshot_input_from_dict(snapshot)?;
nucleide_r2s::snapshot::snapshot_inventory(&input)
.map(|totals| totals.into_iter().collect())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn r2s_expand_sweep(
axes: Vec<BTreeMap<String, Bound<'_, pyo3::types::PyAny>>>,
) -> PyResult<Vec<BTreeMap<String, String>>> {
use pyo3::types::PyAnyMethods;
let mut parsed = Vec::with_capacity(axes.len());
for axis in &axes {
let name: String = axis
.get("name")
.and_then(|v| v.extract().ok())
.ok_or_else(|| PyValueError::new_err("sweep axis needs a `name` string"))?;
let values: Vec<f64> = axis
.get("values")
.and_then(|v| v.extract().ok())
.ok_or_else(|| PyValueError::new_err("sweep axis needs a `values` float list"))?;
parsed.push(
nucleide_r2s::sweep::SweepAxis::new(&name, values)
.map_err(|e| PyValueError::new_err(e.to_string()))?,
);
}
let cases = nucleide_r2s::sweep::expand_sweep(&parsed)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(cases
.into_iter()
.map(|c| {
let mut d = BTreeMap::new();
d.insert("name".to_string(), c.name);
d.insert(
"params".to_string(),
c.params
.iter()
.map(|(k, v)| format!("{k}={v}"))
.collect::<Vec<_>>()
.join(","),
);
d
})
.collect())
}
#[pyfunction]
fn r2s_from_snapshot(
py: Python<'_>,
snapshot: &Bound<'_, pyo3::types::PyDict>,
) -> PyResult<Py<PyAny>> {
let input = snapshot_input_from_dict(snapshot)?;
let (workflow, template, decks) = py
.detach(move || nucleide_r2s::snapshot::snapshot_workflow(&input))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("workflow", r2s_workflow_to_py(py, &workflow))
.ok();
out.set_item("deck", template.to_string()).ok();
let deck_texts: Vec<String> = decks.iter().map(ToString::to_string).collect();
out.set_item("decks", deck_texts).ok();
Ok(out.into_any().unbind())
}
fn parse_integrator(name: &str) -> PyResult<nucleide_depletion::Integrator> {
use nucleide_depletion::Integrator as I;
if name.eq_ignore_ascii_case("predictor") {
return Ok(I::Predictor);
}
if name.eq_ignore_ascii_case("cecm") {
return Ok(I::Cecm);
}
if name.eq_ignore_ascii_case("cf4") {
return Ok(I::Cf4);
}
Err(PyValueError::new_err(format!(
"unsupported integrator `{name}` (supported: predictor, cecm, cf4)"
)))
}
#[pyfunction]
#[pyo3(signature = (chain, n0, dts, rates=None, rates_list=None, integrator="predictor", order=48, method="cram48"))]
#[allow(clippy::too_many_arguments)]
fn deplete_series(
chain: &PyChain,
n0: BTreeMap<String, f64>,
dts: Vec<f64>,
rates: Option<RateMap>,
rates_list: Option<Vec<Option<RateMap>>>,
integrator: &str,
order: u8,
method: &str,
) -> PyResult<Py<PyAny>> {
use nucleide_depletion::{DepletionSystem, ReactionRates, Step};
let integrator = parse_integrator(integrator)?;
let method = resolve_method(order, method)?;
if let Some(list) = &rates_list {
if list.len() != dts.len() {
return Err(PyValueError::new_err(format!(
"rates_list has {} entries but dts has {}",
list.len(),
dts.len()
)));
}
}
if dts.is_empty() {
return Err(PyValueError::new_err("dts must not be empty"));
}
let mut n0_vec = vec![0.0; chain.inner.len()];
for (name, value) in &n0 {
let idx = chain.inner.index_of(name).ok_or_else(|| {
PyValueError::new_err(format!("unknown nuclide `{name}` for this chain"))
})?;
n0_vec[idx] = *value;
}
let empty = BTreeMap::new();
let mut steps = Vec::with_capacity(dts.len());
for (i, dt) in dts.iter().enumerate() {
let step_rates = rates_list
.as_ref()
.and_then(|list| list[i].as_ref())
.or(rates.as_ref())
.unwrap_or(&empty);
let rs = split_rates(step_rates, &chain.inner)?;
steps.push(Step::new(*dt, rs));
}
let template = DepletionSystem::build((*chain.inner).clone(), &ReactionRates::new())
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let series =
nucleide_depletion::integrate_with_method(&template, &n0_vec, &steps, integrator, method)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let names: Vec<&str> = template
.chain
.nuclides
.iter()
.map(|nuc| nuc.name.as_str())
.collect();
let keyed = |rows: &[Vec<f64>]| -> Vec<BTreeMap<String, f64>> {
rows.iter()
.map(|row| {
names
.iter()
.zip(row)
.map(|(name, v)| ((*name).to_string(), *v))
.collect()
})
.collect()
};
let atoms = keyed(&series.atoms[1..]);
let activity = keyed(&series.activity[1..]);
let decay_heat = keyed(&series.decay_heat[1..]);
let times = series.times[1..].to_vec();
Ok(Python::attach(|py| {
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("times", ×).ok();
out.set_item("atoms", &atoms).ok();
out.set_item("activity", &activity).ok();
out.set_item("decay_heat", &decay_heat).ok();
out.into_any().unbind()
}))
}
#[pyfunction]
fn simple_xs(name: &str) -> PyResult<Option<(f64, f64)>> {
NuclideId::from_name(name).map_err(wrap_nucid_err)?;
Ok(nucleide_nuclei::data::simple_xs_by_name(name))
}
#[pyfunction]
fn scattering_length(name: &str) -> PyResult<Option<f64>> {
NuclideId::from_name(name).map_err(wrap_nucid_err)?;
Ok(nucleide_nuclei::data::scattering_length_by_name(name).map(|(b_coh, _)| b_coh))
}
#[pyfunction]
fn decay_energy(name: &str) -> PyResult<Option<f64>> {
NuclideId::from_name(name).map_err(wrap_nucid_err)?;
Ok(nucleide_nuclei::data::decay_energy_mev_by_name(name))
}
#[pyfunction]
fn decay_branches(name: &str) -> PyResult<Vec<(String, f64, String)>> {
NuclideId::from_name(name).map_err(wrap_nucid_err)?;
Ok(nucleide_nuclei::data::decay_branches_by_name(name)
.unwrap_or_default()
.into_iter()
.map(|b| {
(
nucleide_nuclei::NuclideId::from_nucid(b.progeny).to_name(),
b.branching_fraction,
b.mode.as_str().to_string(),
)
})
.collect())
}
#[pyfunction]
fn decay_branch_fraction(parent: &str, progeny: &str) -> PyResult<Option<f64>> {
NuclideId::from_name(parent).map_err(wrap_nucid_err)?;
NuclideId::from_name(progeny).map_err(wrap_nucid_err)?;
Ok(nucleide_nuclei::data::branching_fraction_by_name(
parent, progeny,
))
}
type PyFissionYieldSets = Vec<(f64, Vec<(String, f64, f64)>)>;
#[pyfunction]
#[pyo3(signature = (parent, origin="n", kind="independent"))]
fn fission_yields(parent: &str, origin: &str, kind: &str) -> PyResult<PyFissionYieldSets> {
NuclideId::from_name(parent).map_err(wrap_nucid_err)?;
let origin = nucleide_nuclei::data::FissionYieldOrigin::parse(origin).ok_or_else(|| {
PyValueError::new_err(format!(
"unknown fission-yield origin `{origin}` (expected `n` or `sf`)"
))
})?;
let kind = nucleide_nuclei::data::FissionYieldKind::parse(kind).ok_or_else(|| {
PyValueError::new_err(format!(
"unknown fission-yield kind `{kind}` (expected `independent` or `cumulative`)"
))
})?;
Ok(
nucleide_nuclei::data::fission_yields_by_name(parent, origin, kind)
.unwrap_or_default()
.into_iter()
.map(|set| {
(
set.energy_ev,
set.products
.into_iter()
.map(|p| {
(
nucleide_nuclei::NuclideId::from_nucid(p.progeny).to_name(),
p.yield_fraction,
p.uncertainty,
)
})
.collect(),
)
})
.collect(),
)
}
#[pyfunction]
fn fission_yield(parent: &str, progeny: &str) -> PyResult<Option<f64>> {
NuclideId::from_name(parent).map_err(wrap_nucid_err)?;
NuclideId::from_name(progeny).map_err(wrap_nucid_err)?;
Ok(nucleide_nuclei::data::fission_yield_by_name(
parent, progeny,
))
}
#[pyfunction]
fn normalize_nuclide(name: &str) -> PyResult<String> {
Ok(nucleide_nuclei::dialects::normalize_nuclide_name(name)
.map_err(|e| PyValueError::new_err(e.to_string()))?
.to_name())
}
#[pyfunction]
fn decay_heat(comp: BTreeMap<String, f64>) -> PyResult<f64> {
let mat = comp_to_material(comp)?;
let analytics = nucleide_material::Analytics {
masses: &nucleide_material::Ame2020,
decays: &nucleide_material::ChainDecays,
};
mat.total_decay_heat(&analytics, &nucleide_material::DecayEnergies)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn parse_dose_pathway(s: &str) -> PyResult<nucleide_material::DosePathway> {
nucleide_material::DosePathway::parse(s).ok_or_else(|| {
PyValueError::new_err(format!(
"unknown dose pathway `{s}` (supported: air, soil, ingest, inhale)"
))
})
}
fn parse_dose_source(s: &str) -> PyResult<nucleide_material::DoseSource> {
nucleide_material::DoseSource::parse(s).ok_or_else(|| {
PyValueError::new_err(format!(
"unknown dose source `{s}` (supported: EPA, DOE, GENII)"
))
})
}
#[pyfunction]
#[pyo3(signature = (name, pathway, source="EPA"))]
fn dose_factor(name: &str, pathway: &str, source: &str) -> PyResult<Option<f64>> {
NuclideId::from_name(name).map_err(wrap_nucid_err)?;
let p = parse_dose_pathway(pathway)?;
let s = parse_dose_source(source)?;
Ok(nucleide_nuclei::data::dose_factor_by_name(name, p, s))
}
fn wrap_fgr15_err(e: nucleide_nuclei::fgr15::Error) -> PyErr {
PyValueError::new_err(e.to_string())
}
#[pyfunction]
#[pyo3(signature = (text, expected_rows))]
fn parse_fgr15_table<'py>(
py: Python<'py>,
text: &str,
expected_rows: usize,
) -> PyResult<pyo3::Bound<'py, pyo3::types::PyDict>> {
let table = nucleide_nuclei::fgr15::parse_table(text, expected_rows).map_err(wrap_fgr15_err)?;
let out = pyo3::types::PyDict::new(py);
out.set_item("scenario", table.scenario().as_str())?;
out.set_item("units", table.units())?;
let coefficients = pyo3::types::PyDict::new(py);
for (nucid, row) in table.iter() {
coefficients.set_item(
nucleide_nuclei::fgr15::name_of(NuclideId::from_nucid(nucid)),
row.to_vec(),
)?;
}
out.set_item("coefficients", coefficients)?;
Ok(out)
}
#[pyfunction]
fn fgr15_age_index(age: &str) -> PyResult<usize> {
nucleide_nuclei::fgr15::Fgr15Age::parse(age)
.map(|a| a.index())
.ok_or_else(|| {
PyValueError::new_err(format!(
"unknown FGR 15 age group `{age}` (supported: newborn, 1, 5, 10, 15, adult)"
))
})
}
#[pyfunction]
#[pyo3(signature = (comp, pathway, source="EPA"))]
fn dose_per_g(comp: BTreeMap<String, f64>, pathway: &str, source: &str) -> PyResult<f64> {
let mat = comp_to_material(comp)?;
let analytics = nucleide_material::Analytics {
masses: &nucleide_material::Ame2020,
decays: &nucleide_material::ChainDecays,
};
let p = parse_dose_pathway(pathway)?;
let s = parse_dose_source(source)?;
mat.total_dose_per_g(&analytics, &nucleide_material::DoseFactors, p, s)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[allow(clippy::type_complexity)]
fn separate_material(
comp: BTreeMap<String, f64>,
effs: BTreeMap<String, f64>,
) -> PyResult<(BTreeMap<String, f64>, BTreeMap<String, f64>)> {
let mat = comp_to_material(comp)?;
let mut table = Vec::with_capacity(effs.len());
for (name, eff) in &effs {
let id = NuclideId::from_name(name)
.map_err(|e| PyValueError::new_err(format!("`{name}`: {e}")))?;
table.push((id, *eff));
}
let (product, tails) = mat
.separate(&table)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let named =
|m: nucleide_material::Material| m.comp.iter().map(|(id, g)| (id.to_name(), *g)).collect();
Ok((named(product), named(tails)))
}
#[pyfunction]
fn blend_material(parts: Vec<(BTreeMap<String, f64>, f64)>) -> PyResult<BTreeMap<String, f64>> {
let mats: Vec<nucleide_material::Material> = parts
.iter()
.map(|(comp, _)| comp_to_material(comp.clone()))
.collect::<PyResult<_>>()?;
let refs: Vec<(&nucleide_material::Material, f64)> =
mats.iter().zip(parts.iter().map(|(_, r)| *r)).collect();
let out = nucleide_material::Material::blend(&refs)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(out.comp.iter().map(|(id, g)| (id.to_name(), *g)).collect())
}
#[pyclass(name = "Cusum")]
struct PyCusum {
inner: nucleide_material::Cusum,
}
#[pymethods]
impl PyCusum {
#[new]
#[pyo3(signature = (ref_shift_k=0.5, alarm_h=4.0, startup=10))]
fn new(ref_shift_k: f64, alarm_h: f64, startup: usize) -> PyResult<Self> {
nucleide_material::Cusum::new(ref_shift_k, alarm_h, startup)
.map(|inner| Self { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn update(&mut self, x: f64) -> bool {
self.inner.update(x)
}
fn status(&self) -> bool {
self.inner.status()
}
fn statistic(&self) -> f64 {
self.inner.statistic()
}
fn count(&self) -> usize {
self.inner.count()
}
fn mean(&self) -> f64 {
self.inner.mean()
}
fn variance(&self) -> f64 {
self.inner.variance()
}
fn std(&self) -> f64 {
self.inner.std()
}
fn reset(&mut self) {
self.inner.reset();
}
}
#[pyclass(name = "DeckProblem")]
struct PyDeckProblem {
inner: std::sync::Mutex<nucleide_mcnp_io::problem::DeckProblem>,
}
fn deck_cell_dict(cell: &nucleide_mcnp_io::cell::CellCard) -> BTreeMap<String, String> {
let mut d = BTreeMap::new();
d.insert("num".to_string(), cell.num.to_string());
d.insert("mat".to_string(), cell.mat.to_string());
d.insert(
"dens".to_string(),
cell.dens.map(|v| v.to_string()).unwrap_or_default(),
);
d.insert("geom".to_string(), cell.geom.render());
d.insert("params".to_string(), cell.params.join(" "));
d
}
#[pymethods]
impl PyDeckProblem {
#[staticmethod]
fn loads(text: &str) -> PyResult<Self> {
nucleide_mcnp_io::problem::parse_deck(text)
.map(|inner| Self {
inner: std::sync::Mutex::new(inner),
})
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[getter]
fn message(&self) -> PyResult<String> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.message
.clone())
}
#[getter]
fn title(&self) -> PyResult<String> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.title
.clone())
}
#[getter]
fn cells(&self) -> PyResult<Vec<BTreeMap<String, String>>> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.cells
.iter()
.map(deck_cell_dict)
.collect())
}
#[getter]
fn surfs(&self) -> PyResult<Vec<BTreeMap<String, String>>> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.surfs
.iter()
.map(|s| {
let mut d = BTreeMap::new();
d.insert("num".to_string(), s.num.to_string());
d.insert("reflecting".to_string(), s.reflecting.to_string());
d.insert(
"transform".to_string(),
s.transform.map(|v| v.to_string()).unwrap_or_default(),
);
d.insert(
"periodic".to_string(),
s.periodic.map(|v| v.to_string()).unwrap_or_default(),
);
d.insert("kind".to_string(), s.kind.keyword().to_string());
d.insert(
"coeffs".to_string(),
s.coeffs
.iter()
.map(|v| v.to_string())
.collect::<Vec<_>>()
.join(" "),
);
d
})
.collect())
}
#[getter]
fn material_numbers(&self) -> PyResult<Vec<u32>> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.materials
.iter()
.map(|m| m.number)
.collect())
}
#[getter]
fn data_names(&self) -> PyResult<Vec<String>> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.data
.iter()
.map(|d| d.name.clone())
.collect())
}
fn dumps(&self) -> PyResult<String> {
let guard = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?;
Ok(nucleide_mcnp_io::problem::write_deck(&guard))
}
fn set_cell_density(&self, cell: u32, dens: f64) -> PyResult<()> {
self.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.set_cell_density(cell, dens)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn set_cell_material(&self, cell: u32, mat: u32) -> PyResult<()> {
self.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.set_cell_material(cell, mat)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[getter]
fn mode(&self) -> PyResult<BTreeMap<String, String>> {
let mode = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.mode()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let mut d = BTreeMap::new();
d.insert("particles".to_string(), mode.particles.join(" "));
Ok(d)
}
#[getter]
fn transforms(&self) -> PyResult<Vec<BTreeMap<String, String>>> {
let transforms = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.transforms()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(transforms
.iter()
.map(|t| {
let mut d = BTreeMap::new();
d.insert("number".to_string(), t.number.to_string());
d.insert(
"displacement".to_string(),
t.displacement
.iter()
.map(|v| v.to_string())
.collect::<Vec<_>>()
.join(" "),
);
d.insert(
"rotation".to_string(),
t.rotation
.iter()
.map(|v| v.to_string())
.collect::<Vec<_>>()
.join(" "),
);
d.insert("in_degrees".to_string(), t.is_in_degrees.to_string());
d.insert("main_to_aux".to_string(), t.is_main_to_aux.to_string());
d.insert("hidden".to_string(), t.hidden.to_string());
d
})
.collect())
}
#[getter]
fn universes(&self) -> PyResult<Vec<BTreeMap<String, String>>> {
let universes = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.universes()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(universes
.iter()
.map(|u| {
let mut d = BTreeMap::new();
d.insert("number".to_string(), u.number.to_string());
d.insert(
"cells".to_string(),
u.cells
.iter()
.map(|v| v.to_string())
.collect::<Vec<_>>()
.join(" "),
);
d.insert(
"not_truncated".to_string(),
u.not_truncated
.iter()
.map(|v| v.to_string())
.collect::<Vec<_>>()
.join(" "),
);
d
})
.collect())
}
#[getter]
fn lattices(&self) -> PyResult<Vec<BTreeMap<String, String>>> {
let lattices = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.lattices()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(lattices
.iter()
.map(|l| {
let mut d = BTreeMap::new();
d.insert("cell".to_string(), l.cell.to_string());
d.insert("lattice".to_string(), l.lattice.to_string());
d
})
.collect())
}
#[getter]
fn fills(&self) -> PyResult<Vec<BTreeMap<String, String>>> {
use nucleide_mcnp_io::semantic::{FillTarget, FillTransform};
let fills = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.fills()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(fills
.iter()
.map(|f| {
let mut d = BTreeMap::new();
d.insert("cell".to_string(), f.cell.to_string());
match &f.target {
FillTarget::Single(u) => {
d.insert("kind".to_string(), "single".to_string());
d.insert("universe".to_string(), u.to_string());
d.insert("min_index".to_string(), String::new());
d.insert("max_index".to_string(), String::new());
d.insert("universes".to_string(), String::new());
}
FillTarget::Matrix {
min_index,
max_index,
universes,
} => {
d.insert("kind".to_string(), "matrix".to_string());
d.insert("universe".to_string(), String::new());
d.insert(
"min_index".to_string(),
min_index
.iter()
.map(|v| v.to_string())
.collect::<Vec<_>>()
.join(" "),
);
d.insert(
"max_index".to_string(),
max_index
.iter()
.map(|v| v.to_string())
.collect::<Vec<_>>()
.join(" "),
);
d.insert(
"universes".to_string(),
universes
.iter()
.map(|u| {
u.map(|v| v.to_string()).unwrap_or_else(|| "-".to_string())
})
.collect::<Vec<_>>()
.join(" "),
);
}
}
match &f.transform {
None => {
d.insert("transform".to_string(), String::new());
d.insert("hidden_transform".to_string(), String::new());
}
Some(FillTransform::Reference(n)) => {
d.insert("transform".to_string(), n.to_string());
d.insert("hidden_transform".to_string(), String::new());
}
Some(FillTransform::Hidden(t)) => {
d.insert("transform".to_string(), String::new());
let mut coords: Vec<String> =
t.displacement.iter().map(|v| v.to_string()).collect();
coords.extend(t.rotation.iter().map(|v| v.to_string()));
d.insert("hidden_transform".to_string(), coords.join(" "));
}
}
d.insert("in_degrees".to_string(), f.in_degrees.to_string());
d
})
.collect())
}
#[getter]
fn importances(&self) -> PyResult<Vec<BTreeMap<String, String>>> {
let importances = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.importances()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(importances
.iter()
.map(|v| {
let mut d = BTreeMap::new();
d.insert("cell".to_string(), v.cell.to_string());
d.insert("particle".to_string(), v.particle.clone());
d.insert("value".to_string(), v.value.to_string());
d
})
.collect())
}
#[getter]
fn volumes(&self) -> PyResult<Vec<BTreeMap<String, String>>> {
let volumes = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.volumes()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(volumes
.iter()
.map(|v| {
let mut d = BTreeMap::new();
d.insert("cell".to_string(), v.cell.to_string());
d.insert("volume".to_string(), v.volume.to_string());
d
})
.collect())
}
#[getter]
fn tallies(&self) -> PyResult<Vec<BTreeMap<String, String>>> {
let tallies = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.tallies()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(tallies
.iter()
.map(|t| {
let mut d = BTreeMap::new();
d.insert("number".to_string(), t.number.to_string());
d.insert("type".to_string(), t.tally_type.to_string());
d.insert("particles".to_string(), t.particles.join(","));
d.insert("entries".to_string(), t.entries.join(" "));
d.insert("fm".to_string(), t.fm.clone().unwrap_or_default().join(" "));
d.insert(
"e_bins".to_string(),
t.e_bins.clone().unwrap_or_default().join(" "),
);
d
})
.collect())
}
#[getter]
fn sdef(&self, py: Python<'_>) -> PyResult<Option<Py<PyAny>>> {
let guard = self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?;
let sdef = guard
.sdef()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
sdef.map(|s| sdef_to_py(py, &s)).transpose()
}
fn validate(&self) -> PyResult<()> {
self.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.validate()
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn validation_notes(&self) -> PyResult<Vec<String>> {
Ok(self
.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.validation_notes())
}
fn set_mode(&self, particles: Vec<String>) -> PyResult<()> {
self.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.set_mode(particles)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn set_cell_universe(&self, cell: u32, universe: u32, not_truncated: bool) -> PyResult<()> {
self.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.set_cell_universe(cell, universe, not_truncated)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn set_cell_lattice(&self, cell: u32, lattice: Option<u8>) -> PyResult<()> {
self.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.set_cell_lattice(cell, lattice)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn set_cell_fill(&self, cell: u32, universe: u32) -> PyResult<()> {
self.inner
.lock()
.map_err(|_| PyValueError::new_err("deck lock poisoned"))?
.set_cell_fill(cell, universe)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
}
#[pyfunction]
fn read_deck(path: &str) -> PyResult<PyDeckProblem> {
nucleide_mcnp_io::problem::parse_deck_file(path)
.map(|inner| PyDeckProblem {
inner: std::sync::Mutex::new(inner),
})
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn parse_deck(text: &str) -> PyResult<PyDeckProblem> {
PyDeckProblem::loads(text)
}
fn sdef_to_py(py: Python<'_>, sdef: &nucleide_mcnp_io::sdef::SdefProblem) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let opt3 = |v: &Option<nucleide_mcnp_io::sdef::SdefRef<[f64; 3]>>| {
v.as_ref().map(|r| r.render()).unwrap_or_default()
};
let opt1 = |v: &Option<nucleide_mcnp_io::sdef::SdefRef<f64>>| {
v.as_ref().map(|r| r.render()).unwrap_or_default()
};
let optu = |v: &Option<nucleide_mcnp_io::sdef::SdefRef<u32>>| {
v.as_ref().map(|r| r.render()).unwrap_or_default()
};
let d = PyDict::new(py);
d.set_item("pos", opt3(&sdef.card.pos))?;
d.set_item("cell", optu(&sdef.card.cell))?;
d.set_item("surf", optu(&sdef.card.surf))?;
d.set_item("vec", opt3(&sdef.card.vec))?;
d.set_item("dir", opt1(&sdef.card.dir))?;
d.set_item("axs", opt3(&sdef.card.axs))?;
d.set_item("rad", opt1(&sdef.card.rad))?;
d.set_item("ext", opt1(&sdef.card.ext))?;
d.set_item("erg", opt1(&sdef.card.erg))?;
d.set_item("nrm", opt1(&sdef.card.nrm))?;
d.set_item(
"par",
sdef.card
.par
.as_ref()
.map(|r| r.render())
.unwrap_or_default(),
)?;
d.set_item("wgt", opt1(&sdef.card.wgt))?;
d.set_item("tme", opt1(&sdef.card.tme))?;
d.set_item("ignored", sdef.card.ignored.clone())?;
let dists: Vec<Py<PyAny>> = sdef
.dists
.iter()
.map(|dist| {
let m = PyDict::new(py);
m.set_item("number", dist.number.to_string())?;
m.set_item("si_option", "L")?;
m.set_item("si", dist.si_text())?;
m.set_item(
"sp_option",
dist.sp.as_ref().map(|_| "D").unwrap_or_default(),
)?;
m.set_item("sp", dist.sp_text())?;
m.set_item(
"sb_option",
dist.sb.as_ref().map(|_| "D").unwrap_or_default(),
)?;
m.set_item("sb", dist.sb_text())?;
Ok(m.into_any().unbind())
})
.collect::<PyResult<Vec<_>>>()?;
d.set_item("distributions", dists)?;
d.set_item("card", sdef.emit())?;
Ok(d.into_any().unbind())
}
#[pyfunction]
fn parse_sdef(py: Python<'_>, text: &str) -> PyResult<Py<PyAny>> {
let sdef = nucleide_mcnp_io::sdef::parse_sdef_text(text)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
sdef_to_py(py, &sdef)
}
fn csg_to_openmc_inner(
deck: &nucleide_mcnp_io::problem::DeckProblem,
) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let (xml, table) = nucleide_csg_xlate::deck_csg_to_openmc_xml(deck)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok((
xml,
table
.entries
.into_iter()
.map(|e| {
let mut d = BTreeMap::new();
d.insert("scope".to_string(), e.scope.to_string());
d.insert("target".to_string(), e.target.to_string());
d.insert("action".to_string(), e.action);
d.insert("reason".to_string(), e.reason);
d
})
.collect(),
))
}
#[pyfunction]
fn parse_csg_to_openmc(text: &str) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let deck = nucleide_mcnp_io::problem::parse_deck(text)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
csg_to_openmc_inner(&deck)
}
#[pyfunction]
fn read_csg_to_openmc(path: &str) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let deck = nucleide_mcnp_io::problem::parse_deck_file(path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
csg_to_openmc_inner(&deck)
}
fn csg_to_serpent_inner(
deck: &nucleide_mcnp_io::problem::DeckProblem,
) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let (text, table) = nucleide_csg_xlate::deck_csg_to_serpent_input(deck)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok((
text,
table
.entries
.into_iter()
.map(|e| {
let mut d = BTreeMap::new();
d.insert("scope".to_string(), e.scope.to_string());
d.insert("target".to_string(), e.target.to_string());
d.insert("action".to_string(), e.action);
d.insert("reason".to_string(), e.reason);
d
})
.collect(),
))
}
#[pyfunction]
fn parse_csg_to_serpent(text: &str) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let deck = nucleide_mcnp_io::problem::parse_deck(text)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
csg_to_serpent_inner(&deck)
}
#[pyfunction]
fn read_csg_to_serpent(path: &str) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let deck = nucleide_mcnp_io::problem::parse_deck_file(path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
csg_to_serpent_inner(&deck)
}
fn csg_to_phits_inner(
deck: &nucleide_mcnp_io::problem::DeckProblem,
) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let (text, table) = nucleide_csg_xlate::deck_csg_to_phits_input(deck)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok((
text,
table
.entries
.into_iter()
.map(|e| {
let mut d = BTreeMap::new();
d.insert("scope".to_string(), e.scope.to_string());
d.insert("target".to_string(), e.target.to_string());
d.insert("action".to_string(), e.action);
d.insert("reason".to_string(), e.reason);
d
})
.collect(),
))
}
#[pyfunction]
fn parse_csg_to_phits(text: &str) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let deck = nucleide_mcnp_io::problem::parse_deck(text)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
csg_to_phits_inner(&deck)
}
#[pyfunction]
fn read_csg_to_phits(path: &str) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let deck = nucleide_mcnp_io::problem::parse_deck_file(path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
csg_to_phits_inner(&deck)
}
fn csg_to_gdml_inner(
deck: &nucleide_mcnp_io::problem::DeckProblem,
) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let (xml, table) = nucleide_csg_xlate::deck_csg_to_gdml(deck)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok((
xml,
table
.entries
.into_iter()
.map(|e| {
let mut d = BTreeMap::new();
d.insert("scope".to_string(), e.scope.to_string());
d.insert("target".to_string(), e.target.to_string());
d.insert("action".to_string(), e.action);
d.insert("reason".to_string(), e.reason);
d
})
.collect(),
))
}
#[pyfunction]
fn parse_csg_to_gdml(text: &str) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let deck = nucleide_mcnp_io::problem::parse_deck(text)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
csg_to_gdml_inner(&deck)
}
#[pyfunction]
fn read_csg_to_gdml(path: &str) -> PyResult<(String, Vec<BTreeMap<String, String>>)> {
let deck = nucleide_mcnp_io::problem::parse_deck_file(path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
csg_to_gdml_inner(&deck)
}
#[pyclass(name = "Inventory")]
struct PyInventory {
chain: std::sync::Arc<nucleide_depletion::Chain>,
atoms: BTreeMap<String, f64>,
}
fn inventory_sys(
chain: &nucleide_depletion::Chain,
rates: &RateMap,
) -> PyResult<nucleide_depletion::DepletionSystem> {
let rs = split_rates(rates, chain)?;
nucleide_depletion::DepletionSystem::build(chain.clone(), &rs)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn parse_quantity_unit(unit: &str) -> PyResult<nucleide_depletion::QuantityUnit> {
nucleide_depletion::QuantityUnit::from_str(unit)
.map_err(|e| PyValueError::new_err(format!("{e:?}")))
}
#[pymethods]
impl PyInventory {
#[new]
#[pyo3(signature = (chain, comp, units="atoms"))]
fn new(chain: &PyChain, comp: BTreeMap<String, f64>, units: &str) -> PyResult<Self> {
let unit = parse_quantity_unit(units)?;
let sys = inventory_sys(&chain.inner, &BTreeMap::new())?;
let inv = nucleide_depletion::DecayInventory::from_units(&comp, unit, &sys)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(Self {
chain: chain.inner.clone(),
atoms: inv.atoms,
})
}
fn numbers(&self) -> BTreeMap<String, f64> {
self.atoms.clone()
}
#[pyo3(signature = (dt, time_unit="s", rates=None, order=48, method="cram48"))]
fn decay(
&self,
dt: f64,
time_unit: &str,
rates: Option<RateMap>,
order: u8,
method: &str,
) -> PyResult<Self> {
let method = resolve_method(order, method)?;
let unit = nucleide_depletion::inventory::time_unit_from_str(time_unit)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let seconds = dt * unit.as_seconds();
let empty = BTreeMap::new();
let step_rates = rates.as_ref().unwrap_or(&empty);
let template = inventory_sys(&self.chain, step_rates)?;
let steps = vec![nucleide_depletion::Step::new(
seconds,
split_rates(step_rates, &self.chain)?,
)];
let series = nucleide_depletion::integrate_with_method(
&template,
&chain_vec(&self.chain, &self.atoms)?,
&steps,
nucleide_depletion::Integrator::Predictor,
method,
)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let names: Vec<String> = self.chain.nuclides.iter().map(|n| n.name.clone()).collect();
let atoms = names
.iter()
.zip(series.atoms.last().cloned().unwrap_or_default())
.map(|(n, v)| (n.clone(), v))
.collect();
Ok(Self {
chain: self.chain.clone(),
atoms,
})
}
fn activities(&self, units: &str) -> PyResult<BTreeMap<String, f64>> {
let unit = parse_quantity_unit(units)?;
let sys = inventory_sys(&self.chain, &BTreeMap::new())?;
let inv = nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
};
inv.activities(&sys, unit)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn masses(&self, units: &str) -> PyResult<BTreeMap<String, f64>> {
let unit = parse_quantity_unit(units)?;
let inv = nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
};
inv.masses(unit)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn moles(&self, units: &str) -> PyResult<BTreeMap<String, f64>> {
let unit = parse_quantity_unit(units)?;
let inv = nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
};
inv.moles(unit)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn activity_fractions(&self) -> PyResult<BTreeMap<String, f64>> {
let sys = inventory_sys(&self.chain, &BTreeMap::new())?;
let inv = nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
};
inv.activity_fractions(&sys)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn mass_fractions(&self) -> PyResult<BTreeMap<String, f64>> {
let inv = nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
};
inv.mass_fractions()
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn mole_fractions(&self) -> BTreeMap<String, f64> {
nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
}
.mole_fractions()
}
fn half_lives_readable(&self) -> BTreeMap<String, String> {
nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
}
.half_lives_readable()
}
fn add(&self, other: &Self) -> Self {
let a = nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
};
let b = nucleide_depletion::DecayInventory {
atoms: other.atoms.clone(),
};
Self {
chain: self.chain.clone(),
atoms: a.add(&b).atoms,
}
}
fn sub(&self, other: &Self) -> Self {
let a = nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
};
let b = nucleide_depletion::DecayInventory {
atoms: other.atoms.clone(),
};
Self {
chain: self.chain.clone(),
atoms: a.sub(&b).atoms,
}
}
fn mul(&self, scalar: f64) -> Self {
let a = nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
};
Self {
chain: self.chain.clone(),
atoms: a.mul(scalar).atoms,
}
}
fn div(&self, scalar: f64) -> Self {
let a = nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
};
Self {
chain: self.chain.clone(),
atoms: a.div(scalar).atoms,
}
}
fn to_csv(&self) -> String {
nucleide_depletion::DecayInventory {
atoms: self.atoms.clone(),
}
.to_csv()
}
#[staticmethod]
fn from_csv(chain: &PyChain, text: &str) -> PyResult<Self> {
let inv = nucleide_depletion::DecayInventory::from_csv(text)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
for name in inv.atoms.keys() {
if chain.inner.index_of(name).is_none() {
return Err(PyValueError::new_err(format!(
"unknown nuclide `{name}` for this chain"
)));
}
}
Ok(Self {
chain: chain.inner.clone(),
atoms: inv.atoms,
})
}
}
fn chain_vec(
chain: &nucleide_depletion::Chain,
atoms: &BTreeMap<String, f64>,
) -> PyResult<Vec<f64>> {
let mut vec = vec![0.0; chain.len()];
for (name, value) in atoms {
let idx = chain.index_of(name).ok_or_else(|| {
PyValueError::new_err(format!("unknown nuclide `{name}` for this chain"))
})?;
vec[idx] = *value;
}
Ok(vec)
}
#[pyfunction]
#[pyo3(signature = (chain, n0, dt, rates=None))]
fn cumulative_decays(
chain: &PyChain,
n0: BTreeMap<String, f64>,
dt: f64,
rates: Option<RateMap>,
) -> PyResult<BTreeMap<String, f64>> {
let empty = BTreeMap::new();
let sys = inventory_sys(&chain.inner, rates.as_ref().unwrap_or(&empty))?;
let vec = chain_vec(&chain.inner, &n0)?;
let out = nucleide_depletion::cumulative_decays(&sys, &vec, dt)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(chain
.inner
.nuclides
.iter()
.zip(out)
.map(|(nuc, v)| (nuc.name.clone(), v))
.collect())
}
#[pyfunction]
fn progeny(chain: &PyChain, name: &str) -> Vec<(String, f64, String)> {
nucleide_depletion::progeny(&chain.inner, name)
}
#[pyfunction]
fn branching_fraction(chain: &PyChain, parent: &str, child: &str) -> Option<f64> {
nucleide_depletion::branching_fraction(&chain.inner, parent, child)
}
#[pyfunction]
fn decay_mode(chain: &PyChain, parent: &str, child: &str) -> Option<String> {
nucleide_depletion::decay_mode(&chain.inner, parent, child)
}
#[pyfunction]
fn chain_edges(chain: &PyChain) -> Vec<(String, String, f64, String)> {
nucleide_depletion::chain_edges(&chain.inner)
}
#[pyfunction]
fn armi_to_nucid(name: &str) -> PyResult<PyNuclide> {
nucleide_nuclei::armi::armi_name_to_nucid(name)
.map(|inner| PyNuclide { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn nucid_to_armi(nuclide: &PyNuclide) -> String {
nucleide_nuclei::armi::nucid_to_armi_label(nuclide.inner)
}
#[pyfunction]
fn mcc3_to_nucid(name: &str) -> PyResult<PyNuclide> {
nucleide_nuclei::armi::mcc3_to_nucid(name)
.map(|inner| PyNuclide { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (comp, widths=None))]
fn check_labels(
comp: BTreeMap<String, f64>,
widths: Option<Vec<usize>>,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let mat = comp_to_material(comp)?;
let widths = widths.unwrap_or_else(|| nucleide_material::DEFAULT_WIDTHS.to_vec());
let collisions = nucleide_material::check_labels(&mat, &widths);
Python::attach(|py| {
Ok(collisions
.into_iter()
.map(|c| {
let mut d = BTreeMap::new();
d.insert(
"truncated".to_string(),
c.truncated.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"width".to_string(),
c.width.into_pyobject(py).unwrap().unbind().into_any(),
);
let members: Vec<String> = c.members.iter().map(|id| id.to_name()).collect();
d.insert(
"members".to_string(),
members.into_pyobject(py).unwrap().unbind().into_any(),
);
d
})
.collect())
})
}
#[pyfunction]
fn audit_material(comp: BTreeMap<String, f64>) -> PyResult<Vec<BTreeMap<String, String>>> {
let mat = comp_to_material(comp)?;
Ok(nucleide_material::audit(&mat, &nucleide_material::Ame2020)
.into_iter()
.map(|issue| {
let mut d = BTreeMap::new();
d.insert("kind".to_string(), format!("{:?}", issue.kind));
d.insert("detail".to_string(), issue.detail);
d
})
.collect())
}
#[pyfunction]
#[pyo3(signature = (comp, name, density=None, mcnp_number=1, xs_suffix="80c", serpent_lib="03c", fluka_fid=1, partisn_zone=1))]
#[allow(clippy::too_many_arguments)]
fn emit_cards(
comp: BTreeMap<String, f64>,
name: &str,
density: Option<f64>,
mcnp_number: u32,
xs_suffix: &str,
serpent_lib: &str,
fluka_fid: u32,
partisn_zone: u32,
) -> PyResult<BTreeMap<String, String>> {
let (emitted, _) = emit_drift_inner(
comp,
name,
density,
mcnp_number,
xs_suffix,
serpent_lib,
fluka_fid,
partisn_zone,
)?;
Ok(emitted
.into_iter()
.map(|e| (e.code.to_string(), e.text))
.collect())
}
#[pyfunction]
#[pyo3(signature = (comp, name, density=None, mcnp_number=1, xs_suffix="80c", serpent_lib="03c", fluka_fid=1, partisn_zone=1))]
#[allow(clippy::too_many_arguments)]
fn emit_drift_table(
comp: BTreeMap<String, f64>,
name: &str,
density: Option<f64>,
mcnp_number: u32,
xs_suffix: &str,
serpent_lib: &str,
fluka_fid: u32,
partisn_zone: u32,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let (_, table) = emit_drift_inner(
comp,
name,
density,
mcnp_number,
xs_suffix,
serpent_lib,
fluka_fid,
partisn_zone,
)?;
drift_table_to_py(table)
}
fn drift_table_to_py(
table: nucleide_emit::DriftTable,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
Python::attach(|py| {
Ok(table
.rows
.into_iter()
.map(|r| {
let mut d = BTreeMap::new();
d.insert(
"code".to_string(),
r.code
.to_string()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"mass_in".to_string(),
r.mass_in.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"mass_out".to_string(),
r.mass_out.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"rel_drift".to_string(),
r.rel_drift.into_pyobject(py).unwrap().unbind().into_any(),
);
let dropped: Vec<BTreeMap<String, Py<PyAny>>> = r
.dropped
.into_iter()
.map(|x| {
let mut dd = BTreeMap::new();
dd.insert(
"nuclide".to_string(),
x.id.to_name()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
dd.insert(
"mass".to_string(),
x.mass.into_pyobject(py).unwrap().unbind().into_any(),
);
dd.insert(
"reason".to_string(),
x.reason.into_pyobject(py).unwrap().unbind().into_any(),
);
dd
})
.collect();
d.insert(
"dropped".to_string(),
dropped.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"reparsed".to_string(),
pyo3::types::PyBool::new(py, r.reparsed)
.to_owned()
.into_any()
.unbind(),
);
d
})
.collect())
})
}
#[allow(clippy::too_many_arguments)]
fn emit_drift_inner(
comp: BTreeMap<String, f64>,
name: &str,
density: Option<f64>,
mcnp_number: u32,
xs_suffix: &str,
serpent_lib: &str,
fluka_fid: u32,
partisn_zone: u32,
) -> PyResult<(Vec<nucleide_emit::Emitted>, nucleide_emit::DriftTable)> {
let mut mat = comp_to_material(comp)?;
mat.set_density(density);
emit_drift_with_mat(
mat,
name,
mcnp_number,
xs_suffix,
serpent_lib,
fluka_fid,
partisn_zone,
)
}
#[allow(clippy::too_many_arguments)]
fn emit_drift_with_mat(
mat: nucleide_material::Material,
name: &str,
mcnp_number: u32,
xs_suffix: &str,
serpent_lib: &str,
fluka_fid: u32,
partisn_zone: u32,
) -> PyResult<(Vec<nucleide_emit::Emitted>, nucleide_emit::DriftTable)> {
let mut opts = nucleide_emit::EmitOptions::new(name);
opts.mcnp_number = mcnp_number;
opts.xs_suffix = xs_suffix.to_string();
opts.serpent_lib = serpent_lib.to_string();
opts.fluka_fid = fluka_fid;
opts.partisn_zone = partisn_zone;
nucleide_emit::emit_drift(&mat, &opts).map_err(|e| PyValueError::new_err(e.to_string()))
}
#[allow(clippy::too_many_arguments)]
fn emit_armi_drift_inner(
comp: BTreeMap<String, f64>,
name: &str,
density: Option<f64>,
mcnp_number: u32,
xs_suffix: &str,
serpent_lib: &str,
fluka_fid: u32,
partisn_zone: u32,
) -> PyResult<(Vec<nucleide_emit::Emitted>, nucleide_emit::DriftTable)> {
let mat = nucleide_emit::armi::from_armi_mass_fracs(comp, density)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
emit_drift_with_mat(
mat,
name,
mcnp_number,
xs_suffix,
serpent_lib,
fluka_fid,
partisn_zone,
)
}
#[pyfunction]
#[pyo3(signature = (comp, name, density=None, mcnp_number=1, xs_suffix="80c", serpent_lib="03c", fluka_fid=1, partisn_zone=1))]
#[allow(clippy::too_many_arguments)]
fn emit_armi_cards(
comp: BTreeMap<String, f64>,
name: &str,
density: Option<f64>,
mcnp_number: u32,
xs_suffix: &str,
serpent_lib: &str,
fluka_fid: u32,
partisn_zone: u32,
) -> PyResult<BTreeMap<String, String>> {
let (emitted, _) = emit_armi_drift_inner(
comp,
name,
density,
mcnp_number,
xs_suffix,
serpent_lib,
fluka_fid,
partisn_zone,
)?;
Ok(emitted
.into_iter()
.map(|e| (e.code.to_string(), e.text))
.collect())
}
#[pyfunction]
#[pyo3(signature = (comp, name, density=None, mcnp_number=1, xs_suffix="80c", serpent_lib="03c", fluka_fid=1, partisn_zone=1))]
#[allow(clippy::too_many_arguments)]
fn emit_armi_drift_table(
comp: BTreeMap<String, f64>,
name: &str,
density: Option<f64>,
mcnp_number: u32,
xs_suffix: &str,
serpent_lib: &str,
fluka_fid: u32,
partisn_zone: u32,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let (_, table) = emit_armi_drift_inner(
comp,
name,
density,
mcnp_number,
xs_suffix,
serpent_lib,
fluka_fid,
partisn_zone,
)?;
drift_table_to_py(table)
}
fn parse_reactivity(
spec: &BTreeMap<String, Py<PyAny>>,
py: Python<'_>,
) -> PyResult<nucleide_kinetics::Reactivity> {
use nucleide_kinetics::Reactivity as R;
let kind: String = get_str(spec, py, "kind", "reactivity spec needs a `kind`")?;
let num = |key: &str| -> PyResult<f64> { get_num(spec, py, key) };
let vec = |key: &str| -> PyResult<Vec<f64>> { get_vec(spec, py, key) };
let r = match kind.as_str() {
"constant" => R::Constant { rho: num("rho")? },
"step" => R::Step {
t_step: num("t_step")?,
rho_init: num("rho_init")?,
rho_final: num("rho_final")?,
},
"impulse" => R::Impulse {
t_start: num("t_start")?,
t_end: num("t_end")?,
rho_init: num("rho_init")?,
rho_max: num("rho_max")?,
},
"ramp" => R::Ramp {
t_start: num("t_start")?,
t_end: num("t_end")?,
rho_init: num("rho_init")?,
rho_rise: num("rho_rise")?,
rho_final: num("rho_final")?,
},
"polyline" => R::Polyline {
times: vec("times")?,
values: vec("values")?,
},
other => {
return Err(PyValueError::new_err(format!(
"unknown reactivity kind `{other}` (supported: constant, step, impulse, ramp, polyline)"
)))
}
};
r.validate()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(r)
}
fn get_str(
spec: &BTreeMap<String, Py<PyAny>>,
py: Python<'_>,
key: &str,
missing: &str,
) -> PyResult<String> {
spec.get(key)
.ok_or_else(|| PyValueError::new_err(missing.to_string()))?
.extract::<String>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a string")))
}
fn get_num(spec: &BTreeMap<String, Py<PyAny>>, py: Python<'_>, key: &str) -> PyResult<f64> {
spec.get(key)
.ok_or_else(|| PyValueError::new_err(format!("reactivity spec missing `{key}`")))?
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a number")))
}
fn get_vec(spec: &BTreeMap<String, Py<PyAny>>, py: Python<'_>, key: &str) -> PyResult<Vec<f64>> {
spec.get(key)
.ok_or_else(|| PyValueError::new_err(format!("reactivity spec missing `{key}`")))?
.extract::<Vec<f64>>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a list of numbers")))
}
fn kinetics_params(
betas: Vec<f64>,
lambdas: Vec<f64>,
lambda_gen: f64,
) -> PyResult<nucleide_kinetics::KineticParams> {
nucleide_kinetics::KineticParams::new(betas, lambdas, lambda_gen)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (betas, lambdas, lambda_gen, rho, t, n0, c0=None, method="trapezoidal", rtol=1e-9, atol=1e-12, dt_min=1e-14, dt_max=None, max_steps=1000000))]
#[allow(clippy::too_many_arguments)]
fn kinetics_solve(
py: Python<'_>,
betas: Vec<f64>,
lambdas: Vec<f64>,
lambda_gen: f64,
rho: BTreeMap<String, Py<PyAny>>,
t: Vec<f64>,
n0: f64,
c0: Option<Vec<f64>>,
method: &str,
rtol: f64,
atol: f64,
dt_min: f64,
dt_max: Option<f64>,
max_steps: usize,
) -> PyResult<Py<PyAny>> {
use nucleide_kinetics::{Method as M, SolverOptions};
let params = kinetics_params(betas, lambdas, lambda_gen)?;
let rho = parse_reactivity(&rho, py)?;
let grid =
nucleide_kinetics::TimeGrid::new(t).map_err(|e| PyValueError::new_err(e.to_string()))?;
let state = nucleide_kinetics::State::new(¶ms, n0, c0)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let method = if method.eq_ignore_ascii_case("trapezoidal") {
M::Trapezoidal
} else if method.eq_ignore_ascii_case("backward_euler") {
M::BackwardEuler
} else {
return Err(PyValueError::new_err(format!(
"unknown kinetics method `{method}` (supported: trapezoidal, backward_euler)"
)));
};
let opts = SolverOptions {
method,
rtol,
atol,
dt_min,
dt_max: dt_max.unwrap_or(f64::INFINITY),
max_steps,
};
let sol = nucleide_kinetics::solve(¶ms, &rho, &grid, &state, &opts)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("times", &sol.times).ok();
out.set_item("n", &sol.n).ok();
out.set_item("C", &sol.c).ok();
out.set_item("n0", sol.initial.n0).ok();
out.set_item("C0", &sol.initial.c0).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
fn kinetics_equilibrium(
betas: Vec<f64>,
lambdas: Vec<f64>,
lambda_gen: f64,
n0: f64,
) -> PyResult<Vec<f64>> {
kinetics_params(betas, lambdas, lambda_gen)?
.equilibrium_precursors(n0)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (betas, lambdas, lambda_gen, rho, n0, c0=None))]
fn kinetics_initial_rate(
py: Python<'_>,
betas: Vec<f64>,
lambdas: Vec<f64>,
lambda_gen: f64,
rho: BTreeMap<String, Py<PyAny>>,
n0: f64,
c0: Option<Vec<f64>>,
) -> PyResult<f64> {
let params = kinetics_params(betas, lambdas, lambda_gen)?;
let rho = parse_reactivity(&rho, py)?;
let state = nucleide_kinetics::State::new(¶ms, n0, c0)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(nucleide_kinetics::solve::initial_rate(
¶ms, &rho, &state,
))
}
#[pyfunction]
fn kinetics_inhour_rho(
betas: Vec<f64>,
lambdas: Vec<f64>,
lambda_gen: f64,
omega: f64,
) -> PyResult<f64> {
let params = kinetics_params(betas, lambdas, lambda_gen)?;
nucleide_kinetics::rho_of_omega(¶ms, omega)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn kinetics_stable_period(
betas: Vec<f64>,
lambdas: Vec<f64>,
lambda_gen: f64,
rho: f64,
) -> PyResult<f64> {
let params = kinetics_params(betas, lambdas, lambda_gen)?;
nucleide_kinetics::stable_period(¶ms, rho).map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn kinetics_prompt_jump(
n_before: f64,
rho_before: f64,
rho_after: f64,
beta_total: f64,
) -> PyResult<f64> {
nucleide_kinetics::prompt_jump(n_before, rho_before, rho_after, beta_total)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (response, rates, guess, tolerance=1e-3, max_iterations=200))]
fn unfold_sandii(
py: Python<'_>,
response: Vec<Vec<f64>>,
rates: Vec<f64>,
guess: Vec<f64>,
tolerance: f64,
max_iterations: usize,
) -> PyResult<Py<PyAny>> {
let sol = nucleide_unfold::sandii::unfold(&response, &rates, &guess, tolerance, max_iterations)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("spectrum", &sol.spectrum).ok();
out.set_item("rates", &sol.rates).ok();
out.set_item("rate_factors", &sol.rate_factors).ok();
out.set_item("iterations", sol.iterations).ok();
out.set_item("tolerance", sol.tolerance).ok();
out.set_item("max_rel_change", sol.max_rel_change).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
fn unfold_forward_fold(response: Vec<Vec<f64>>, spectrum: Vec<f64>) -> PyResult<Vec<f64>> {
nucleide_unfold::forward_fold(&response, &spectrum)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn parse_plasma_reaction(name: &str) -> PyResult<nucleide_plasma_source::FusionReaction> {
use nucleide_plasma_source::FusionReaction as R;
match name
.to_ascii_lowercase()
.replace(['-', '_', ' '], "")
.as_str()
{
"dt" => Ok(R::Dt),
"dd" => Ok(R::Dd),
other => Err(PyValueError::new_err(format!(
"unknown fusion reaction `{other}` (supported: dt, dd)"
))),
}
}
enum PyPlasmaSource {
Basic(nucleide_plasma_source::PlasmaSourceConfig),
Parametric(nucleide_plasma_source::ParametricPlasmaConfig),
}
fn parse_plasma_basic_spec(
spec: &BTreeMap<String, Py<PyAny>>,
py: Python<'_>,
kind: &str,
) -> PyResult<nucleide_plasma_source::PlasmaSourceConfig> {
use nucleide_plasma_source as ps;
let num = |key: &str| -> PyResult<f64> {
spec.get(key)
.ok_or_else(|| PyValueError::new_err(format!("source spec missing `{key}`")))?
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a number")))
};
let reaction = parse_plasma_reaction(&get_str(
spec,
py,
"reaction",
"source spec missing `reaction`",
)?)?;
let model = match kind {
"point" => {
let position: Vec<f64> = spec
.get("position")
.ok_or_else(|| PyValueError::new_err("point source needs `position` [cm]"))?
.extract::<Vec<f64>>(py)
.map_err(|_| PyValueError::new_err("`position` must be a list of numbers"))?;
if position.len() != 3 {
return Err(PyValueError::new_err(
"`position` must have exactly three entries",
));
}
ps::SourceModel::Point(ps::PointSource {
x_cm: position[0],
y_cm: position[1],
z_cm: position[2],
})
}
"ring" => ps::SourceModel::Ring(ps::RingSource {
radius_cm: num("radius")?,
height_cm: num("height")?,
}),
other => {
return Err(PyValueError::new_err(format!(
"unknown source kind `{other}` (supported: point, ring, parametric)"
)))
}
};
let mut config = ps::PlasmaSourceConfig {
model,
reaction,
ion_temperature_kev: num("ion_temperature_kev")?,
weight: 1.0,
};
if let Some(weight) = spec.get("weight") {
let weight = weight
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err("`weight` must be a number"))?;
config = config.with_weight(weight);
}
config
.validate()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(config)
}
fn parse_plasma_parametric_spec(
spec: &BTreeMap<String, Py<PyAny>>,
py: Python<'_>,
) -> PyResult<nucleide_plasma_source::ParametricPlasmaConfig> {
use nucleide_plasma_source as ps;
let num = |key: &str| -> PyResult<f64> {
spec.get(key)
.ok_or_else(|| PyValueError::new_err(format!("parametric spec missing `{key}`")))?
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a number")))
};
for key in ["fuel", "start_angle", "rotation_angle"] {
if spec.contains_key(key) {
return Err(PyValueError::new_err(format!(
"plasma-source: not yet supported: `{key}` (fuel mixtures are \
Eriksson-weighted reactant distributions; sectors need a \
toroidal-angle distribution — both outside the parametric model)"
)));
}
}
let mode = ps::ProfileMode::parse(&get_str(
spec,
py,
"mode",
"parametric spec missing `mode`",
)?)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let fuel = parse_plasma_reaction(&get_str(
spec,
py,
"reaction",
"parametric spec missing `reaction`",
)?)?;
let mut config = ps::ParametricPlasmaConfig {
geometry: ps::MillerGeometry {
major_radius_cm: num("major_radius")?,
minor_radius_cm: num("minor_radius")?,
elongation: num("elongation")?,
triangularity: num("triangularity")?,
shafranov_factor_cm: num("shafranov_factor")?,
},
mode,
ion_density: ps::DensityProfile {
centre_m3: num("ion_density_centre")?,
peaking_factor: num("ion_density_peaking_factor")?,
pedestal_m3: num("ion_density_pedestal")?,
separatrix_m3: num("ion_density_separatrix")?,
},
ion_temperature: ps::TemperatureProfile {
centre_kev: num("ion_temperature_centre")?,
peaking_factor: num("ion_temperature_peaking_factor")?,
beta: num("ion_temperature_beta")?,
pedestal_kev: num("ion_temperature_pedestal")?,
separatrix_kev: num("ion_temperature_separatrix")?,
},
pedestal_radius_cm: num("pedestal_radius")?,
fuel,
weight: 1.0,
};
if let Some(weight) = spec.get("weight") {
let weight = weight
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err("`weight` must be a number"))?;
config.weight = weight;
}
config
.validate()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(config)
}
fn parse_plasma_source_spec(
spec: &BTreeMap<String, Py<PyAny>>,
py: Python<'_>,
) -> PyResult<PyPlasmaSource> {
let kind: String = spec
.get("kind")
.ok_or_else(|| PyValueError::new_err("source spec needs a `kind`"))?
.extract::<String>(py)
.map_err(|_| PyValueError::new_err("`kind` must be a string"))?;
match kind.as_str() {
"point" | "ring" => Ok(PyPlasmaSource::Basic(parse_plasma_basic_spec(
spec, py, &kind,
)?)),
"parametric" => Ok(PyPlasmaSource::Parametric(parse_plasma_parametric_spec(
spec, py,
)?)),
other => Err(PyValueError::new_err(format!(
"unknown source kind `{other}` (supported: point, ring, parametric)"
))),
}
}
fn plasma_drift_rows(
py: Python<'_>,
report: &nucleide_plasma_source::DriftReport,
) -> PyResult<Vec<Py<pyo3::types::PyDict>>> {
use pyo3::types::PyDict;
let mut rows = Vec::with_capacity(report.rows.len());
for row in &report.rows {
let d = PyDict::new(py);
d.set_item("quantity", &row.quantity)?;
d.set_item("accounted", row.accounted)?;
d.set_item("rel_drift", row.rel_drift)?;
d.set_item("reparsed", row.reparsed)?;
d.set_item("note", &row.note)?;
rows.push(d.unbind());
}
Ok(rows)
}
#[pyfunction]
#[pyo3(signature = (spec, n, seed))]
fn plasma_source_particles(
py: Python<'_>,
spec: BTreeMap<String, Py<PyAny>>,
n: usize,
seed: u64,
) -> PyResult<Py<PyAny>> {
use nucleide_plasma_source as ps;
let particles = match parse_plasma_source_spec(&spec, py)? {
PyPlasmaSource::Basic(config) => ps::SourceSampler::new(config, seed)
.map_err(|e| PyValueError::new_err(e.to_string()))?
.sample_n(n),
PyPlasmaSource::Parametric(config) => ps::ParametricSampler::new(config, seed)
.map_err(|e| PyValueError::new_err(e.to_string()))?
.sample_n(n),
};
let mut x = Vec::with_capacity(n);
let mut y = Vec::with_capacity(n);
let mut z = Vec::with_capacity(n);
let mut u = Vec::with_capacity(n);
let mut v = Vec::with_capacity(n);
let mut w = Vec::with_capacity(n);
let mut energy = Vec::with_capacity(n);
let mut weight = Vec::with_capacity(n);
for p in &particles {
x.push(p.position_cm[0]);
y.push(p.position_cm[1]);
z.push(p.position_cm[2]);
u.push(p.direction[0]);
v.push(p.direction[1]);
w.push(p.direction[2]);
energy.push(p.energy_mev);
weight.push(p.weight);
}
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("x", x.into_pyarray(py))?;
out.set_item("y", y.into_pyarray(py))?;
out.set_item("z", z.into_pyarray(py))?;
out.set_item("u", u.into_pyarray(py))?;
out.set_item("v", v.into_pyarray(py))?;
out.set_item("w", w.into_pyarray(py))?;
out.set_item("energy", energy.into_pyarray(py))?;
out.set_item("weight", weight.into_pyarray(py))?;
Ok(out.into_any().unbind())
}
#[pyfunction]
#[pyo3(signature = (spec, bins=21))]
fn plasma_source_emit_cards(
py: Python<'_>,
spec: BTreeMap<String, Py<PyAny>>,
bins: usize,
) -> PyResult<Py<PyAny>> {
use nucleide_plasma_source as ps;
let source = parse_plasma_source_spec(&spec, py)?;
let version = match spec.get("mcnp_version") {
Some(v) => v
.extract::<u32>(py)
.map_err(|_| PyValueError::new_err("`mcnp_version` must be an integer (5 or 6)"))?,
None => 5,
};
let emit = |card: ps::EmittedCard| -> PyResult<Py<pyo3::types::PyDict>> {
use pyo3::types::PyDict;
let d = PyDict::new(py);
d.set_item("card", card.text)?;
d.set_item("drift", plasma_drift_rows(py, &card.drift)?)?;
Ok(d.unbind())
};
use pyo3::types::PyDict;
let out = PyDict::new(py);
let (nominal, mean, sigma, mono) = match &source {
PyPlasmaSource::Basic(config) => {
let sdef = ps::emit_sdef(config, version, bins)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let serpent =
ps::emit_serpent(config, bins).map_err(|e| PyValueError::new_err(e.to_string()))?;
let spectrum = config
.spectrum()
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let sigma = match spectrum {
ps::SpectrumSpec::Gaussian { sigma_mev, .. } => sigma_mev,
ps::SpectrumSpec::Mono { .. } => 0.0,
};
out.set_item("sdef", emit(sdef)?)?;
out.set_item("serpent", emit(serpent)?)?;
(
config.reaction.nominal_energy_mev(),
spectrum.mean_mev(),
sigma,
spectrum.is_mono(),
)
}
PyPlasmaSource::Parametric(config) => {
let sdef = ps::emit_sdef_parametric(config, version, bins)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let serpent = ps::emit_serpent_parametric(config, bins)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let (mean, sigma) = config
.fuel
.moments_mev(config.temperature_kev(0.0))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
out.set_item("sdef", emit(sdef)?)?;
out.set_item("serpent", emit(serpent)?)?;
(config.fuel.nominal_energy_mev(), mean, sigma, sigma == 0.0)
}
};
let spec_out = PyDict::new(py);
spec_out.set_item("nominal_mev", nominal)?;
spec_out.set_item("mean_mev", mean)?;
spec_out.set_item("sigma_mev", sigma)?;
spec_out.set_item("mono", mono)?;
out.set_item("spectrum", spec_out)?;
Ok(out.into_any().unbind())
}
#[pyfunction]
fn plasma_source_spectrum_moments(
py: Python<'_>,
reaction: &str,
ion_temperature_kev: f64,
) -> PyResult<Py<PyAny>> {
use nucleide_plasma_source::FusionReaction as R;
let reaction = parse_plasma_reaction(reaction)?;
let (mean, sigma) = reaction
.moments_mev(ion_temperature_kev)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item(
"reaction",
match reaction {
R::Dt => "dt",
R::Dd => "dd",
},
)?;
out.set_item("label", reaction.label())?;
out.set_item("nominal_mev", reaction.nominal_energy_mev())?;
out.set_item("mean_mev", mean)?;
out.set_item("sigma_mev", sigma)?;
Ok(out.into_any().unbind())
}
#[pyfunction]
fn plasma_source_reactivity(reaction: &str, ion_temperature_kev: f64) -> PyResult<f64> {
parse_plasma_reaction(reaction)?
.reactivity_m3_per_s(ion_temperature_kev)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn parse_damage_nuclide(key: &Bound<'_, PyAny>) -> PyResult<NuclideId> {
if let Ok(nucid) = key.extract::<u32>() {
return NuclideId::try_from_nucid(nucid).map_err(wrap_nucid_err);
}
if let Ok(name) = key.extract::<&str>() {
return NuclideId::from_name(name).map_err(wrap_nucid_err);
}
Err(PyTypeError::new_err("expected int nucid or str name"))
}
#[pyfunction]
#[pyo3(signature = (flux, response, bounds, seconds))]
fn damage_nrt_dpa(
flux: Vec<f64>,
response: Vec<f64>,
bounds: Vec<f64>,
seconds: f64,
) -> PyResult<f64> {
nucleide_damage::nrt_dpa(&flux, &response, &bounds, seconds)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (flux, response, bounds, seconds))]
fn damage_arc_dpa(
flux: Vec<f64>,
response: Vec<f64>,
bounds: Vec<f64>,
seconds: f64,
) -> PyResult<f64> {
nucleide_damage::arc_dpa(&flux, &response, &bounds, seconds)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (flux, response, bounds, seconds))]
fn damage_gas_appm(
flux: Vec<f64>,
response: Vec<f64>,
bounds: Vec<f64>,
seconds: f64,
) -> PyResult<f64> {
nucleide_damage::gas_appm(&flux, &response, &bounds, seconds)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (flux, he_response, damage_response, bounds, seconds))]
fn damage_he_dpa_ratio(
flux: Vec<f64>,
he_response: Vec<f64>,
damage_response: Vec<f64>,
bounds: Vec<f64>,
seconds: f64,
) -> PyResult<f64> {
nucleide_damage::he_dpa_ratio(&flux, &he_response, &damage_response, &bounds, seconds)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (t_ev, recoil, lattice))]
fn damage_lindhard_partition(
t_ev: f64,
recoil: &Bound<'_, PyAny>,
lattice: &Bound<'_, PyAny>,
) -> PyResult<f64> {
let recoil = parse_damage_nuclide(recoil)?;
let lattice = parse_damage_nuclide(lattice)?;
nucleide_damage::lindhard_partition(t_ev, &recoil, &lattice)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (t_ev, recoil, lattice))]
fn damage_damage_energy(
t_ev: f64,
recoil: &Bound<'_, PyAny>,
lattice: &Bound<'_, PyAny>,
) -> PyResult<f64> {
let recoil = parse_damage_nuclide(recoil)?;
let lattice = parse_damage_nuclide(lattice)?;
nucleide_damage::damage_energy(t_ev, &recoil, &lattice)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (t_ev, ed_ev, target))]
fn damage_nrt_displacements(t_ev: f64, ed_ev: f64, target: &Bound<'_, PyAny>) -> PyResult<f64> {
let target = parse_damage_nuclide(target)?;
nucleide_damage::nrt_displacements(t_ev, ed_ev, &target)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (t_dam_ev, ed_ev, b_arc, c_arc))]
fn damage_arc_efficiency(t_dam_ev: f64, ed_ev: f64, b_arc: f64, c_arc: f64) -> PyResult<f64> {
let params = nucleide_damage::ArcParams::new(b_arc, c_arc)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
nucleide_damage::arc_efficiency(t_dam_ev, ed_ev, ¶ms)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (t_ev, ed_ev, target, b_arc, c_arc))]
fn damage_arc_displacements(
t_ev: f64,
ed_ev: f64,
target: &Bound<'_, PyAny>,
b_arc: f64,
c_arc: f64,
) -> PyResult<f64> {
let target = parse_damage_nuclide(target)?;
let params = nucleide_damage::ArcParams::new(b_arc, c_arc)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
nucleide_damage::arc_displacements(t_ev, ed_ev, &target, ¶ms)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (metric, flux, response, bounds, seconds, mean, cov, n, seed, k))]
#[allow(clippy::too_many_arguments)] fn damage_fold_uq(
py: Python<'_>,
metric: &str,
flux: Vec<f64>,
response: Vec<f64>,
bounds: Vec<f64>,
seconds: f64,
mean: Vec<f64>,
cov: Vec<Vec<f64>>,
n: usize,
seed: u64,
k: f64,
) -> PyResult<Py<PyAny>> {
use nucleide_damage::FoldMetric as M;
let metric = match metric
.to_ascii_lowercase()
.replace(['-', ' '], "_")
.as_str()
{
"nrt_dpa" => M::NrtDpa,
"arc_dpa" => M::ArcDpa,
"gas_appm" => M::GasAppm,
"he_dpa_ratio" => M::HeDpaRatio,
other => {
return Err(PyValueError::new_err(format!(
"unknown fold metric `{other}` (supported: nrt_dpa, arc_dpa, gas_appm)"
)))
}
};
let s = nucleide_damage::fold_uq(
metric, &flux, &response, &bounds, seconds, &mean, &cov, n, seed, k,
)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("metric", s.metric.name())?;
out.set_item("nominal", s.nominal)?;
out.set_item("mean", s.mean)?;
out.set_item("std", s.std)?;
out.set_item("expected", s.expected)?;
out.set_item("analytic_std", s.analytic_std)?;
out.set_item("k", s.k)?;
out.set_item("n", s.n)?;
out.set_item("seed", s.seed)?;
out.set_item("passed", s.passed)?;
Ok(out.into_any().unbind())
}
fn parse_tritium_boundary(
spec: &BTreeMap<String, Py<PyAny>>,
py: Python<'_>,
) -> PyResult<nucleide_tritium::Boundary> {
use nucleide_tritium::Boundary as B;
let kind: String = spec
.get("kind")
.ok_or_else(|| PyValueError::new_err("boundary spec needs a `kind`"))?
.extract::<String>(py)
.map_err(|_| PyValueError::new_err("`kind` must be a string"))?;
let num = |key: &str| -> PyResult<f64> {
spec.get(key)
.ok_or_else(|| PyValueError::new_err(format!("boundary spec missing `{key}`")))?
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a number")))
};
let b = match kind.as_str() {
"dirichlet" => B::dirichlet(num("value")?),
"sieverts" => B::sieverts(num("solubility")?, num("pressure")?),
"henry" => B::henry(num("solubility")?, num("pressure")?),
"recombination" => B::recombination(num("rate")?),
"zero_flux" => Ok(B::ZeroFlux),
other => {
return Err(PyValueError::new_err(format!(
"unknown boundary kind `{other}` (supported: dirichlet, sieverts, henry, recombination, zero_flux)"
)))
}
};
b.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn parse_tritium_trap(
spec: &BTreeMap<String, Py<PyAny>>,
py: Python<'_>,
) -> PyResult<nucleide_tritium::TrapSpec> {
let num = |key: &str| -> PyResult<f64> {
spec.get(key)
.ok_or_else(|| PyValueError::new_err(format!("trap spec missing `{key}`")))?
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a number")))
};
let opt = |key: &str| -> PyResult<f64> {
match spec.get(key) {
None => Ok(0.0),
Some(v) => v
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a number"))),
}
};
nucleide_tritium::TrapSpec::new(
num("k0")?,
opt("e_k")?,
num("p0")?,
opt("e_p")?,
num("site_density")?,
)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[allow(clippy::too_many_arguments)]
fn tritium_params(
py: Python<'_>,
length: f64,
cells: usize,
d0: f64,
e_d: f64,
traps: Vec<BTreeMap<String, Py<PyAny>>>,
temperature: Vec<f64>,
source: Option<Vec<f64>>,
) -> PyResult<nucleide_tritium::TransportParams> {
let parsed: Vec<nucleide_tritium::TrapSpec> = traps
.iter()
.map(|s| parse_tritium_trap(s, py))
.collect::<PyResult<_>>()?;
nucleide_tritium::TransportParams::new(
length,
cells,
d0,
e_d,
parsed,
temperature,
source.unwrap_or_default(),
)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (length, cells, d0, e_d, traps, temperature, source, left, right))]
#[allow(clippy::too_many_arguments)]
fn tritium_steady(
py: Python<'_>,
length: f64,
cells: usize,
d0: f64,
e_d: f64,
traps: Vec<BTreeMap<String, Py<PyAny>>>,
temperature: Vec<f64>,
source: Option<Vec<f64>>,
left: BTreeMap<String, Py<PyAny>>,
right: BTreeMap<String, Py<PyAny>>,
) -> PyResult<Py<PyAny>> {
let params = tritium_params(py, length, cells, d0, e_d, traps, temperature, source)?;
let left = parse_tritium_boundary(&left, py)?;
let right = parse_tritium_boundary(&right, py)?;
let s = nucleide_tritium::steady_state(¶ms, &left, &right)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("centres", &s.centres).ok();
out.set_item("mobile", &s.mobile).ok();
out.set_item("trapped", &s.trapped).ok();
out.set_item("flux_left", s.flux_left).ok();
out.set_item("flux_right", s.flux_right).ok();
out.set_item("inventory_mobile", s.inventory_mobile).ok();
out.set_item("inventory_trapped", s.inventory_trapped).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
#[pyo3(signature = (length, cells, d0, e_d, traps, temperature, source, left, right, t, mobile0=None, trapped0=None, method="crank_nicolson", rtol=1e-9, atol=1e-12, dt_min=1e-14, dt_max=None, max_steps=1000000))]
#[allow(clippy::too_many_arguments)]
fn tritium_transient(
py: Python<'_>,
length: f64,
cells: usize,
d0: f64,
e_d: f64,
traps: Vec<BTreeMap<String, Py<PyAny>>>,
temperature: Vec<f64>,
source: Option<Vec<f64>>,
left: BTreeMap<String, Py<PyAny>>,
right: BTreeMap<String, Py<PyAny>>,
t: Vec<f64>,
mobile0: Option<Vec<f64>>,
trapped0: Option<Vec<Vec<f64>>>,
method: &str,
rtol: f64,
atol: f64,
dt_min: f64,
dt_max: Option<f64>,
max_steps: usize,
) -> PyResult<Py<PyAny>> {
use nucleide_tritium::{SolverOptions, Theta};
let params = tritium_params(py, length, cells, d0, e_d, traps, temperature, source)?;
let left = parse_tritium_boundary(&left, py)?;
let right = parse_tritium_boundary(&right, py)?;
let grid =
nucleide_tritium::TimeGrid::new(t).map_err(|e| PyValueError::new_err(e.to_string()))?;
let ntraps = params.traps.len();
let mobile = mobile0.unwrap_or_else(|| vec![0.0; params.cells]);
let trapped = trapped0.unwrap_or_else(|| vec![vec![0.0; ntraps]; params.cells]);
let initial = nucleide_tritium::InitialState::new(¶ms, mobile, trapped)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let theta = if method.eq_ignore_ascii_case("crank_nicolson") {
Theta::CrankNicolson
} else if method.eq_ignore_ascii_case("backward_euler") {
Theta::BackwardEuler
} else {
return Err(PyValueError::new_err(format!(
"unknown tritium method `{method}` (supported: crank_nicolson, backward_euler)"
)));
};
let opts = SolverOptions {
theta,
rtol,
atol,
dt_min,
dt_max: dt_max.unwrap_or(f64::INFINITY),
max_steps,
};
let sol = nucleide_tritium::solve(¶ms, &left, &right, &grid, &initial, &opts)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("times", &sol.times).ok();
out.set_item("mobile", &sol.mobile).ok();
out.set_item("trapped", &sol.trapped).ok();
out.set_item("flux_left", &sol.flux_left).ok();
out.set_item("flux_right", &sol.flux_right).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
fn tritium_time_lag(length: f64, diffusivity: f64) -> PyResult<f64> {
nucleide_tritium::time_lag(length, diffusivity)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn tritium_breakthrough(diffusivity: f64, length: f64, times: Vec<f64>) -> PyResult<Vec<f64>> {
times
.iter()
.map(|t| {
nucleide_tritium::breakthrough_ratio(diffusivity, length, *t)
.map_err(|e| PyValueError::new_err(e.to_string()))
})
.collect()
}
#[pyfunction]
fn tritium_oriani(diffusivity: f64, equilibrium_constant: f64, site_density: f64) -> PyResult<f64> {
nucleide_tritium::effective_diffusivity(diffusivity, equilibrium_constant, site_density)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn tritium_langmuir(site_density: f64, equilibrium_constant: f64, c_mobile: f64) -> PyResult<f64> {
nucleide_tritium::equilibrium_trapped(site_density, equilibrium_constant, c_mobile)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn tritium_irreversible_fill(
rate_k: f64,
c_mobile: f64,
site_density: f64,
times: Vec<f64>,
) -> PyResult<Vec<f64>> {
times
.iter()
.map(|t| {
nucleide_tritium::irreversible_fill(rate_k, c_mobile, site_density, *t)
.map_err(|e| PyValueError::new_err(e.to_string()))
})
.collect()
}
#[pyfunction]
fn tritium_sieverts(solubility: f64, pressure: f64) -> PyResult<f64> {
nucleide_tritium::sieverts_concentration(solubility, pressure)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn tritium_recombination_rate(kr0: f64, e_r: f64, temp: f64) -> PyResult<f64> {
nucleide_tritium::recombination_rate_arrhenius(kr0, e_r, temp)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn parse_tritium_layer(
spec: &BTreeMap<String, Py<PyAny>>,
py: Python<'_>,
) -> PyResult<nucleide_tritium::LayerSpec> {
let num = |key: &str| -> PyResult<f64> {
spec.get(key)
.ok_or_else(|| PyValueError::new_err(format!("layer spec missing `{key}`")))?
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a number")))
};
let opt = |key: &str| -> PyResult<f64> {
match spec.get(key) {
None => Ok(0.0),
Some(v) => v
.extract::<f64>(py)
.map_err(|_| PyValueError::new_err(format!("`{key}` must be a number"))),
}
};
let traps = match spec.get("traps") {
None => Vec::new(),
Some(v) => v
.extract::<Vec<BTreeMap<String, Py<PyAny>>>>(py)
.map_err(|_| PyValueError::new_err("`traps` must be a list of dicts"))?
.iter()
.map(|s| parse_tritium_trap(s, py))
.collect::<PyResult<_>>()?,
};
let temperature: Vec<f64> = match spec.get("temperature") {
None => vec![500.0],
Some(v) => v
.extract::<Vec<f64>>(py)
.map_err(|_| PyValueError::new_err("`temperature` must be a list of numbers"))?,
};
let source: Vec<f64> = match spec.get("source") {
None => Vec::new(),
Some(v) => v
.extract::<Vec<f64>>(py)
.map_err(|_| PyValueError::new_err("`source` must be a list of numbers"))?,
};
let cells: usize = spec
.get("cells")
.ok_or_else(|| PyValueError::new_err("layer spec missing `cells`"))?
.extract::<usize>(py)
.map_err(|_| PyValueError::new_err("`cells` must be an integer"))?;
nucleide_tritium::LayerSpec::new(
num("thickness")?,
cells,
num("D")?,
opt("E_D")?,
num("solubility")?,
traps,
temperature,
source,
)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn tritium_layer_stack(
py: Python<'_>,
layers: Vec<BTreeMap<String, Py<PyAny>>>,
) -> PyResult<nucleide_tritium::LayerStack> {
let parsed: Vec<nucleide_tritium::LayerSpec> = layers
.iter()
.map(|s| parse_tritium_layer(s, py))
.collect::<PyResult<_>>()?;
let interfaces = vec![nucleide_tritium::Interface::Sieverts; parsed.len().saturating_sub(1)];
nucleide_tritium::LayerStack::new(parsed, interfaces)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn tritium_layers_steady(
py: Python<'_>,
layers: Vec<BTreeMap<String, Py<PyAny>>>,
left: BTreeMap<String, Py<PyAny>>,
right: BTreeMap<String, Py<PyAny>>,
) -> PyResult<Py<PyAny>> {
let stack = tritium_layer_stack(py, layers)?;
let left = parse_tritium_boundary(&left, py)?;
let right = parse_tritium_boundary(&right, py)?;
let s = nucleide_tritium::steady_layers(&stack, &left, &right)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("centres", &s.centres).ok();
out.set_item("mobile", &s.mobile).ok();
out.set_item("trapped", &s.trapped).ok();
out.set_item("flux_left", s.flux_left).ok();
out.set_item("flux_right", s.flux_right).ok();
out.set_item("inventory_mobile", s.inventory_mobile).ok();
out.set_item("inventory_trapped", s.inventory_trapped).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
#[pyo3(signature = (layers, left, right, t, mobile0=None, trapped0=None, method="crank_nicolson", rtol=1e-9, atol=1e-12, dt_min=1e-14, dt_max=None, max_steps=1000000))]
#[allow(clippy::too_many_arguments)]
fn tritium_layers_transient(
py: Python<'_>,
layers: Vec<BTreeMap<String, Py<PyAny>>>,
left: BTreeMap<String, Py<PyAny>>,
right: BTreeMap<String, Py<PyAny>>,
t: Vec<f64>,
mobile0: Option<Vec<f64>>,
trapped0: Option<Vec<Vec<f64>>>,
method: &str,
rtol: f64,
atol: f64,
dt_min: f64,
dt_max: Option<f64>,
max_steps: usize,
) -> PyResult<Py<PyAny>> {
use nucleide_tritium::{SolverOptions, Theta};
let stack = tritium_layer_stack(py, layers)?;
let left = parse_tritium_boundary(&left, py)?;
let right = parse_tritium_boundary(&right, py)?;
let grid =
nucleide_tritium::TimeGrid::new(t).map_err(|e| PyValueError::new_err(e.to_string()))?;
let mobile = mobile0.unwrap_or_else(|| vec![0.0; stack.total_cells()]);
let trapped = trapped0.unwrap_or_else(|| stack.zero_state().trapped);
let initial = nucleide_tritium::InitialState { mobile, trapped };
let theta = if method.eq_ignore_ascii_case("crank_nicolson") {
Theta::CrankNicolson
} else if method.eq_ignore_ascii_case("backward_euler") {
Theta::BackwardEuler
} else {
return Err(PyValueError::new_err(format!(
"unknown tritium method `{method}` (supported: crank_nicolson, backward_euler)"
)));
};
let opts = SolverOptions {
theta,
rtol,
atol,
dt_min,
dt_max: dt_max.unwrap_or(f64::INFINITY),
max_steps,
};
let sol = nucleide_tritium::solve_layers(&stack, &left, &right, &grid, &initial, &opts)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
use pyo3::types::PyDict;
let out = PyDict::new(py);
out.set_item("times", &sol.times).ok();
out.set_item("mobile", &sol.mobile).ok();
out.set_item("trapped", &sol.trapped).ok();
out.set_item("flux_left", &sol.flux_left).ok();
out.set_item("flux_right", &sol.flux_right).ok();
Ok(out.into_any().unbind())
}
#[pyfunction]
fn spectroscopy_rect_smooth(counts: Vec<f64>, m: i64) -> PyResult<Vec<f64>> {
let w = usize::try_from(m).map_err(|_| {
PyValueError::new_err(format!("spectroscopy: smoothing width {m} is less than 3"))
})?;
nucleide_spectroscopy::rect_smooth(&counts, w).map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn spectroscopy_five_point_smooth(counts: Vec<f64>) -> PyResult<Vec<f64>> {
nucleide_spectroscopy::five_point_smooth(&counts)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn spectroscopy_calc_bg(
counts: Vec<f64>,
channels: Vec<f64>,
c1: i64,
c2: i64,
m: i64,
) -> PyResult<f64> {
nucleide_spectroscopy::calc_bg(&counts, &channels, c1, c2, m)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn spectroscopy_gross_count(
counts: Vec<f64>,
channels: Vec<f64>,
c1: i64,
c2: i64,
) -> PyResult<f64> {
nucleide_spectroscopy::gross_count(&counts, &channels, c1, c2)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn spectroscopy_net_counts(
counts: Vec<f64>,
channels: Vec<f64>,
c1: i64,
c2: i64,
m: i64,
) -> PyResult<f64> {
nucleide_spectroscopy::net_counts(&counts, &channels, c1, c2, m)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn spectroscopy_energy_bins(channels: Vec<f64>, calib_e_fit: Vec<f64>) -> PyResult<Vec<f64>> {
nucleide_spectroscopy::energy_bins(&channels, &calib_e_fit)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn spectroscopy_detector_efficiency(
energy_mev: f64,
eff_coeff: Vec<f64>,
eff_fit: i64,
) -> PyResult<f64> {
nucleide_spectroscopy::detector_efficiency(energy_mev, &eff_coeff, eff_fit)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (energies, effs, weights, order, eff_fit=1))]
fn spectroscopy_fit_efficiency(
energies: Vec<f64>,
effs: Vec<f64>,
weights: Vec<f64>,
order: usize,
eff_fit: i64,
) -> PyResult<Vec<f64>> {
nucleide_spectroscopy::fit_efficiency(&energies, &effs, &weights, order, eff_fit)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn atomic_key(atomic: &BTreeMap<String, f64>, key: &str) -> PyResult<f64> {
atomic
.get(key)
.copied()
.ok_or_else(|| PyValueError::new_err(format!("atomic constants missing `{key}`")))
}
#[pyfunction]
#[pyo3(signature = (atomic, k_conv=None, l_conv=None))]
fn spectroscopy_xray_lines(
atomic: BTreeMap<String, f64>,
k_conv: Option<f64>,
l_conv: Option<f64>,
) -> PyResult<Vec<(f64, f64)>> {
let data = nucleide_spectroscopy::AtomicData {
k_shell_fluor: atomic_key(&atomic, "k_shell_fluor")?,
l_shell_fluor: atomic_key(&atomic, "l_shell_fluor")?,
prob: atomic_key(&atomic, "prob")?,
kb_to_ka: atomic_key(&atomic, "kb_to_ka")?,
ka2_to_ka1: atomic_key(&atomic, "ka2_to_ka1")?,
ka1_en_kev: atomic_key(&atomic, "ka1_en_kev")?,
ka2_en_kev: atomic_key(&atomic, "ka2_en_kev")?,
kb_en_kev: atomic_key(&atomic, "kb_en_kev")?,
l_en_kev: atomic_key(&atomic, "l_en_kev")?,
};
let present = |v: Option<f64>| v.filter(|x| !x.is_nan());
Ok(
nucleide_spectroscopy::xray_lines(&data, present(k_conv), present(l_conv))
.iter()
.map(|l| (l.energy_kev, l.intensity))
.collect(),
)
}
#[pyfunction]
#[pyo3(signature = (lines, x=0.0, y=0.0, z=0.0, u=0.0, v=0.0, w=0.0, weight=1.0, particle="Neutron", version=5))]
#[allow(clippy::too_many_arguments)]
fn spectroscopy_sdef_decay_source(
lines: Vec<(f64, f64)>,
x: f64,
y: f64,
z: f64,
u: f64,
v: f64,
w: f64,
weight: f64,
particle: &str,
version: u32,
) -> PyResult<(Vec<(f64, f64)>, String)> {
let particle = particle
.parse::<nucleide_nuclei::particles::ParticleId>()
.map_err(|e| PyValueError::new_err(format!("spectroscopy: particle {e}")))?;
let source = nucleide_spectroscopy::PointSource {
x,
y,
z,
u,
v,
w,
weight,
particle,
};
nucleide_spectroscopy::sdef_card(&lines, &source, version)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn spectrum_to_py(
py: Python<'_>,
spec: &nucleide_spectroscopy::GammaSpectrum,
) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let d = PyDict::new(py);
let s = &spec.spectrum;
d.set_item("spec_name", &s.spec_name)?;
d.set_item("start_chan_num", s.start_chan_num)?;
d.set_item("num_channels", s.num_channels)?;
d.set_item("channels", &s.channels)?;
d.set_item("counts", &s.counts)?;
d.set_item("ebin", &s.ebin)?;
d.set_item("real_time", spec.real_time)?;
d.set_item("live_time", spec.live_time)?;
d.set_item("dead_time", spec.dead_time())?;
d.set_item("det_id", &spec.det_id)?;
d.set_item("det_descp", &spec.det_descp)?;
d.set_item("start_date", &spec.start_date)?;
d.set_item("start_time", &spec.start_time)?;
d.set_item("calib_e_fit", &spec.calib_e_fit)?;
d.set_item("calib_fwhm_fit", &spec.calib_fwhm_fit)?;
d.set_item("file_name", &spec.file_name)?;
Ok(d.into_any().unbind())
}
#[pyfunction]
fn spectroscopy_parse_dollar_spe(py: Python<'_>, text: &str) -> PyResult<Py<PyAny>> {
let spec = nucleide_spectroscopy::parse_dollar_spe(text, "")
.map_err(|e| PyValueError::new_err(e.to_string()))?;
spectrum_to_py(py, &spec)
}
#[pyfunction]
fn spectroscopy_parse_spe(py: Python<'_>, text: &str) -> PyResult<Py<PyAny>> {
let spec = nucleide_spectroscopy::parse_plain_spe(text, "")
.map_err(|e| PyValueError::new_err(e.to_string()))?;
spectrum_to_py(py, &spec)
}
#[pyfunction]
fn spectroscopy_read_dollar_spe(py: Python<'_>, path: &str) -> PyResult<Py<PyAny>> {
let text = std::fs::read_to_string(path).map_err(|e| PyValueError::new_err(e.to_string()))?;
let spec = nucleide_spectroscopy::parse_dollar_spe(&text, path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
spectrum_to_py(py, &spec)
}
#[pyfunction]
fn spectroscopy_read_spe(py: Python<'_>, path: &str) -> PyResult<Py<PyAny>> {
let text = std::fs::read_to_string(path).map_err(|e| PyValueError::new_err(e.to_string()))?;
let spec = nucleide_spectroscopy::parse_plain_spe(&text, path)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
spectrum_to_py(py, &spec)
}
#[pyfunction]
fn spectroscopy_parse_lines_tsv(text: &str) -> PyResult<Vec<(f64, f64)>> {
nucleide_spectroscopy::parse_lines_tsv(text).map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn spectroscopy_read_decay_lines(path: &str) -> PyResult<Vec<(f64, f64)>> {
let text = std::fs::read_to_string(path).map_err(|e| PyValueError::new_err(e.to_string()))?;
nucleide_spectroscopy::parse_lines_tsv(&text).map_err(|e| PyValueError::new_err(e.to_string()))
}
fn uq_sample_err(e: nucleide_linalg::sample::SampleError) -> PyErr {
PyValueError::new_err(e.to_string())
}
fn uq_decay_err(e: nucleide_linalg::decay::DecayError) -> PyErr {
PyValueError::new_err(e.to_string())
}
#[pyfunction]
fn uq_sample_mvn(
py: Python<'_>,
mean: Vec<f64>,
cov: Vec<Vec<f64>>,
n: usize,
seed: u64,
) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let set = nucleide_linalg::sample::sample_mvn(&mean, &cov, n, seed).map_err(uq_sample_err)?;
let d = PyDict::new(py);
d.set_item("samples", set.samples)?;
d.set_item("method", set.method.name())?;
match &set.method {
nucleide_linalg::sample::FactorMethod::Cholesky => {
d.set_item("min_eigen", py.None())?;
d.set_item("max_eigen", py.None())?;
}
nucleide_linalg::sample::FactorMethod::EigenClip {
min_eigen,
max_eigen,
} => {
d.set_item("min_eigen", *min_eigen)?;
d.set_item("max_eigen", *max_eigen)?;
}
}
Ok(d.into_any().unbind())
}
#[pyfunction]
fn uq_sample_mean(samples: Vec<Vec<f64>>) -> PyResult<Vec<f64>> {
nucleide_linalg::sample::sample_mean(&samples).map_err(uq_sample_err)
}
#[pyfunction]
fn uq_sample_cov(samples: Vec<Vec<f64>>) -> PyResult<Vec<Vec<f64>>> {
nucleide_linalg::sample::sample_cov(&samples).map_err(uq_sample_err)
}
#[pyfunction]
fn uq_check_convergence(
py: Python<'_>,
mean: Vec<f64>,
cov: Vec<Vec<f64>>,
samples: Vec<Vec<f64>>,
mean_tol: f64,
cov_tol: f64,
) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let rep = nucleide_linalg::sample::check_convergence(&mean, &cov, &samples, mean_tol, cov_tol)
.map_err(uq_sample_err)?;
let d = PyDict::new(py);
d.set_item("mean_err_max", rep.mean_err_max)?;
d.set_item("cov_err_fro", rep.cov_err_fro)?;
d.set_item("mean_tol", rep.mean_tol)?;
d.set_item("cov_tol", rep.cov_tol)?;
d.set_item("passed", rep.passed)?;
Ok(d.into_any().unbind())
}
#[pyfunction]
fn uq_perturb_branches(base: Vec<f64>, rel: Vec<f64>) -> PyResult<Vec<f64>> {
nucleide_linalg::decay::perturb_branches(&base, &rel).map_err(uq_decay_err)
}
#[pyfunction]
fn uq_perturb_energies(base: Vec<f64>, delta: Vec<f64>, convention: &str) -> PyResult<Vec<f64>> {
let conv = nucleide_linalg::sample::PerturbConvention::parse(convention)
.map_err(PyValueError::new_err)?;
nucleide_linalg::decay::perturb_energies(&base, &delta, conv).map_err(uq_decay_err)
}
#[pyfunction]
fn uq_sample_lognormal(
py: Python<'_>,
mean_log: Vec<f64>,
cov: Vec<Vec<f64>>,
n: usize,
seed: u64,
) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let set = nucleide_linalg::sample::sample_lognormal(&mean_log, &cov, n, seed)
.map_err(uq_sample_err)?;
let d = PyDict::new(py);
d.set_item("samples", set.samples)?;
d.set_item("method", set.method.name())?;
match &set.method {
nucleide_linalg::sample::FactorMethod::Cholesky => {
d.set_item("min_eigen", py.None())?;
d.set_item("max_eigen", py.None())?;
}
nucleide_linalg::sample::FactorMethod::EigenClip {
min_eigen,
max_eigen,
} => {
d.set_item("min_eigen", *min_eigen)?;
d.set_item("max_eigen", *max_eigen)?;
}
}
Ok(d.into_any().unbind())
}
#[pyfunction]
fn uq_sample_lhs(
py: Python<'_>,
mean: Vec<f64>,
cov: Vec<Vec<f64>>,
n: usize,
seed: u64,
) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let set = nucleide_linalg::sample::sample_lhs(&mean, &cov, n, seed).map_err(uq_sample_err)?;
let d = PyDict::new(py);
d.set_item("samples", set.samples)?;
d.set_item("method", set.method.name())?;
match &set.method {
nucleide_linalg::sample::FactorMethod::Cholesky => {
d.set_item("min_eigen", py.None())?;
d.set_item("max_eigen", py.None())?;
}
nucleide_linalg::sample::FactorMethod::EigenClip {
min_eigen,
max_eigen,
} => {
d.set_item("min_eigen", *min_eigen)?;
d.set_item("max_eigen", *max_eigen)?;
}
}
Ok(d.into_any().unbind())
}
#[pyfunction]
fn uq_lognormal_mean(mean_log: Vec<f64>, cov: Vec<Vec<f64>>) -> PyResult<Vec<f64>> {
nucleide_linalg::sample::lognormal_mean(&mean_log, &cov).map_err(uq_sample_err)
}
#[pyfunction]
fn uq_lognormal_cov(mean_log: Vec<f64>, cov: Vec<Vec<f64>>) -> PyResult<Vec<Vec<f64>>> {
nucleide_linalg::sample::lognormal_cov(&mean_log, &cov).map_err(uq_sample_err)
}
#[pyfunction]
fn uq_passthrough(delta: Vec<f64>) -> PyResult<Vec<f64>> {
nucleide_linalg::decay::passthrough(&delta).map_err(uq_decay_err)
}
#[pyfunction]
fn uq_perturb_fission_yields(base: Vec<f64>, rel: Vec<f64>) -> PyResult<Vec<f64>> {
nucleide_linalg::decay::perturb_fission_yields(&base, &rel).map_err(uq_decay_err)
}
fn parse_projectile(flag: &str) -> PyResult<nucleide_nuclei::rxname::Projectile> {
flag.parse::<nucleide_nuclei::rxname::Projectile>()
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn resolve_rx_id(spec: &Bound<'_, PyAny>) -> PyResult<u32> {
if let Ok(id) = spec.extract::<u32>() {
return Ok(id);
}
if let Ok(s) = spec.extract::<&str>() {
return nucleide_nuclei::rxname::name_to_id(s)
.map_err(|e| PyValueError::new_err(e.to_string()));
}
Err(PyTypeError::new_err(
"expected reaction id (int) or name (str)",
))
}
#[pyfunction]
fn rxname_label(id: u32) -> &'static str {
nucleide_nuclei::rxname::label(id)
}
#[pyfunction]
fn rxname_doc(id: u32) -> &'static str {
nucleide_nuclei::rxname::doc(id)
}
#[pyfunction]
fn rxname_reaction(py: Python<'_>, id: u32) -> PyResult<Option<Py<PyAny>>> {
use pyo3::types::PyDict;
Ok(nucleide_nuclei::rxname::reaction(id).map(|r| {
let d = PyDict::new(py);
d.set_item("id", r.id).ok();
d.set_item("name", r.name).ok();
d.set_item("mt", r.mt).ok();
d.set_item("label", r.label).ok();
d.set_item("doc", r.doc).ok();
d.into_any().unbind()
}))
}
#[pyfunction]
#[pyo3(signature = (from_nucid, to_nucid, projectile="n"))]
fn rxname_id_from_nucdelta(from_nucid: u32, to_nucid: u32, projectile: &str) -> PyResult<u32> {
let p = parse_projectile(projectile)?;
nucleide_nuclei::rxname::id_from_nucdelta(from_nucid, to_nucid, p)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (parent, rx, projectile="n"))]
fn rxname_child(parent: &str, rx: &Bound<'_, PyAny>, projectile: &str) -> PyResult<String> {
let p = parse_projectile(projectile)?;
let rx = resolve_rx_id(rx)?;
let parent_id = NuclideId::from_name(parent)
.map_err(|e| PyValueError::new_err(format!("`{parent}`: {e}")))?;
nucleide_nuclei::rxname::child(parent_id, rx, p)
.map(|id| id.to_name())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (child, rx, projectile="n"))]
fn rxname_parent(child: &str, rx: &Bound<'_, PyAny>, projectile: &str) -> PyResult<String> {
let p = parse_projectile(projectile)?;
let rx = resolve_rx_id(rx)?;
let child_id = NuclideId::from_name(child)
.map_err(|e| PyValueError::new_err(format!("`{child}`: {e}")))?;
nucleide_nuclei::rxname::parent(child_id, rx, p)
.map(|id| id.to_name())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn particle_is_valid(spec: &str) -> bool {
nucleide_nuclei::particles::is_valid(spec)
}
#[pyfunction]
fn particle_is_valid_pdc(n: i32) -> bool {
nucleide_nuclei::particles::is_valid_pdc(n)
}
#[pyfunction]
fn particle_is_hydrogen(spec: &str) -> bool {
nucleide_nuclei::particles::is_hydrogen(spec)
}
#[pyfunction]
fn particle_is_heavy_ion(spec: &str) -> bool {
nucleide_nuclei::particles::is_heavy_ion(spec)
}
#[pyfunction]
#[pyo3(signature = (name, source="EPA"))]
fn dose_f1(name: &str, source: &str) -> PyResult<Option<f64>> {
NuclideId::from_name(name).map_err(wrap_nucid_err)?;
let s = parse_dose_source(source)?;
Ok(nucleide_nuclei::data::dose_f1_by_name(name, s))
}
#[pyfunction]
#[pyo3(signature = (name, source="EPA"))]
fn dose_lung_model(name: &str, source: &str) -> PyResult<Option<char>> {
NuclideId::from_name(name).map_err(wrap_nucid_err)?;
let s = parse_dose_source(source)?;
Ok(nucleide_nuclei::data::dose_lung_model_by_name(name, s))
}
fn bare_element_z(name: &str) -> Option<u32> {
let t = name.trim();
if t.is_empty() {
return None;
}
let mut chars = t.chars();
let first = chars.next()?.to_uppercase().next()?;
let rest: String = chars.collect::<String>().to_lowercase();
let canon = format!("{first}{rest}");
nucleide_nuclei::element_z(&canon)
}
fn mat_from_comp_elements(comp: BTreeMap<String, f64>) -> PyResult<nucleide_material::Material> {
let mut mat = nucleide_material::Material::new();
for (name, grams) in &comp {
let id = match NuclideId::from_name(name) {
Ok(id) => id,
Err(_) => match bare_element_z(name) {
Some(z) => NuclideId::from_nucid(z * 10_000_000),
None => {
return Err(PyValueError::new_err(format!(
"`{name}`: unknown nuclide or element"
)));
}
},
};
mat.add_nuclide(id, *grams);
}
Ok(mat)
}
fn mat_to_comp_elements(mat: &nucleide_material::Material) -> BTreeMap<String, f64> {
let mut out = BTreeMap::new();
for (&id, &grams) in &mat.comp {
let key = if id.a() == 0 && id.state() == 0 {
nucleide_nuclei::element_symbol(id.z())
.unwrap_or("X")
.to_string()
} else {
id.to_name()
};
*out.entry(key).or_insert(0.0) += grams;
}
out
}
#[pyfunction]
fn mix_by_mass(parts: Vec<(BTreeMap<String, f64>, f64)>) -> PyResult<BTreeMap<String, f64>> {
let mats: Vec<nucleide_material::Material> = parts
.iter()
.map(|(comp, _)| mat_from_comp_elements(comp.clone()))
.collect::<PyResult<_>>()?;
let refs: Vec<(&nucleide_material::Material, f64)> =
mats.iter().zip(parts.iter().map(|(_, w)| *w)).collect();
let out = nucleide_material::Material::mix_by_mass(&refs)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(mat_to_comp_elements(&out))
}
#[pyfunction]
fn mix_by_volume(parts: Vec<(BTreeMap<String, f64>, f64, f64)>) -> PyResult<BTreeMap<String, f64>> {
let mut mats: Vec<nucleide_material::Material> = Vec::with_capacity(parts.len());
for (comp, _, density) in &parts {
let mut m = mat_from_comp_elements(comp.clone())?;
m.set_density(Some(*density));
mats.push(m);
}
let refs: Vec<(&nucleide_material::Material, f64)> =
mats.iter().zip(parts.iter().map(|(_, v, _)| *v)).collect();
let out = nucleide_material::Material::mix_by_volume(&refs)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(mat_to_comp_elements(&out))
}
#[pyfunction]
fn specific_activity(comp: BTreeMap<String, f64>) -> PyResult<f64> {
let mat = mat_from_comp_elements(comp)?;
let analytics = nucleide_material::Analytics {
masses: &nucleide_material::Ame2020,
decays: &nucleide_material::ChainDecays,
};
mat.specific_activity(&analytics)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (entries, cross_sections=None))]
fn materials_doc_to_xml(
entries: Vec<(String, BTreeMap<String, f64>, f64)>,
cross_sections: Option<String>,
) -> PyResult<String> {
let mut doc = nucleide_material::MaterialsDoc::new();
if let Some(path) = cross_sections {
doc = doc.cross_sections(path);
}
for (name, comp, density) in entries {
let mut mat = mat_from_comp_elements(comp)?;
mat.set_density(Some(density));
doc = doc.push(name, mat);
}
doc.to_xml()
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn expand_elements(comp: BTreeMap<String, f64>) -> PyResult<BTreeMap<String, f64>> {
let mut mat = mat_from_comp_elements(comp)?;
mat.expand_elements(
&nucleide_material::Ame2020,
&nucleide_material::NaturalAbundances,
)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(mat_to_comp_elements(&mat))
}
#[pyfunction]
fn collapse_elements(comp: BTreeMap<String, f64>) -> PyResult<BTreeMap<String, f64>> {
let mat = mat_from_comp_elements(comp)?;
Ok(mat_to_comp_elements(&mat.collapse_elements()))
}
fn parse_fluka_nuc(spec: &str) -> PyResult<nucleide_fluka_io::material::FlukaNuc> {
use nucleide_fluka_io::material::FlukaNuc;
if let Ok(id) = NuclideId::from_name(spec) {
return Ok(FlukaNuc::Nuclide(id));
}
if let Some(z) = bare_element_z(spec) {
return Ok(FlukaNuc::Element(z));
}
if let Ok(z) = spec.trim().parse::<u32>() {
if nucleide_nuclei::element_symbol(z).is_some() {
return Ok(FlukaNuc::Element(z));
}
}
Err(PyValueError::new_err(format!(
"`{spec}`: unknown nuclide or element"
)))
}
#[pyfunction]
fn fluka_material_str(fid: u32, nuc: &str, density: f64) -> PyResult<String> {
let parsed = parse_fluka_nuc(nuc)?;
nucleide_fluka_io::material::material_str(fid, parsed, density)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (fid, compound_name, density, frac_type="mass", components=None))]
fn fluka_compound_str(
fid: u32,
compound_name: &str,
density: f64,
frac_type: &str,
components: Option<Vec<(String, f64)>>,
) -> PyResult<String> {
use nucleide_fluka_io::material::{Component, FracType};
let frac = match frac_type.trim().to_ascii_lowercase().as_str() {
"mass" => FracType::Mass,
"atom" => FracType::Atom,
other => {
return Err(PyValueError::new_err(format!(
"frac_type must be mass|atom, got `{other}`"
)));
}
};
let pairs = components.unwrap_or_default();
let comps: Vec<Component> = pairs
.iter()
.map(|(nuc, frac)| parse_fluka_nuc(nuc).map(|n| Component::new(n, *frac)))
.collect::<PyResult<_>>()?;
nucleide_fluka_io::material::compound_str(fid, compound_name, density, frac, &comps)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn fluka_builtin_set() -> Vec<String> {
let mut out: Vec<String> = nucleide_fluka_io::material::builtin_set()
.into_iter()
.map(str::to_string)
.collect();
out.sort();
out
}
#[pyfunction]
fn alara_validate_deck(text: &str) -> PyResult<()> {
let deck = nucleide_alara_io::AlaraDeck::parse(text)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
deck.validate()
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn alara_check_block(block: &str, line: usize) -> PyResult<()> {
nucleide_alara_io::AlaraDeck::check_block(block, line)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn alara_flux_total(name: &str, text: &str) -> PyResult<f64> {
nucleide_alara_io::FluxSpec::parse(name, text)
.map(|f| f.total())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn alara_flux_len(name: &str, text: &str) -> PyResult<usize> {
nucleide_alara_io::FluxSpec::parse(name, text)
.map(|f| f.len())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn alara_output_totals(
py: Python<'_>,
text: &str,
run_lbl: &str,
) -> PyResult<Vec<BTreeMap<String, Py<PyAny>>>> {
let owned_text = text.to_owned();
let owned_lbl = run_lbl.to_owned();
let frame = py
.detach(move || {
nucleide_alara_io::output::ResponseFrame::parse(&owned_text, &owned_lbl)
.map(|f| f.totals())
})
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(frame
.rows
.iter()
.map(|r| fispact_row_to_map(py, r))
.collect())
}
#[pyfunction]
fn alara_output_total_activity(text: &str, run_lbl: &str) -> PyResult<f64> {
nucleide_alara_io::output::ResponseFrame::parse(text, run_lbl)
.map(|f| f.total_activity())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn alara_photon_total_strength(text: &str) -> PyResult<f64> {
nucleide_alara_io::PhotonSource::from_str(text)
.map(|p| p.total_strength())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (deck_text, top=None))]
fn alara_schedule_total_time(deck_text: &str, top: Option<&str>) -> PyResult<f64> {
let owned = deck_text.to_owned();
let owned_top = top.map(str::to_owned);
let steps =
expand_deck_schedules(&owned, owned_top.as_deref()).map_err(PyValueError::new_err)?;
Ok(nucleide_alara_io::schedule::total_time(&steps))
}
#[pyfunction]
fn alara_clearance_eu_table() -> BTreeMap<String, f64> {
nucleide_alara_io::ClearanceTable::eu_annex_vii()
.iter()
.map(|(nuc, limit)| (nucleide_nuclei::dialects::serpent(nuc), limit))
.collect()
}
fn clearance_key(key: &str) -> PyResult<NuclideId> {
nucleide_nuclei::dialects::normalize_nuclide_name(key)
.map_err(|e| PyValueError::new_err(format!("bad nuclide name `{key}`: {e}")))
}
fn clearance_pairs(map: &BTreeMap<String, f64>, what: &str) -> PyResult<Vec<(NuclideId, f64)>> {
map.iter()
.map(|(name, value)| Ok((clearance_key(name)?, *value)))
.collect::<PyResult<_>>()
.map_err(|e| PyValueError::new_err(format!("{what}: {e}")))
}
fn clearance_table_from(
map: &BTreeMap<String, f64>,
) -> PyResult<nucleide_alara_io::ClearanceTable> {
let mut table = nucleide_alara_io::ClearanceTable::new();
for (nuc, limit) in clearance_pairs(map, "limits")? {
table
.insert(nuc, limit)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
}
Ok(table)
}
#[pyfunction]
#[pyo3(signature = (inventory, limits=None))]
fn alara_clearance_index(
inventory: BTreeMap<String, f64>,
limits: Option<BTreeMap<String, f64>>,
) -> PyResult<f64> {
let table = match limits {
Some(map) => clearance_table_from(&map)?,
None => nucleide_alara_io::ClearanceTable::eu_annex_vii(),
};
let pairs = clearance_pairs(&inventory, "inventory")?;
nucleide_alara_io::clearance_index(&pairs, &table)
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (inventory, limits=None))]
fn alara_sum_of_fractions(
py: Python<'_>,
inventory: BTreeMap<String, f64>,
limits: Option<BTreeMap<String, f64>>,
) -> PyResult<BTreeMap<String, Py<PyAny>>> {
let table = match limits {
Some(map) => clearance_table_from(&map)?,
None => nucleide_alara_io::ClearanceTable::eu_annex_vii(),
};
let pairs = clearance_pairs(&inventory, "inventory")?;
let out = nucleide_alara_io::sum_of_fractions(&pairs, &table)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let mut d = BTreeMap::new();
d.insert(
"sum".to_string(),
out.sum.into_pyobject(py).unwrap().unbind().into_any(),
);
d.insert(
"class".to_string(),
out.class
.to_string()
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"max_fraction".to_string(),
out.max_fraction
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
d.insert(
"max_nuclide".to_string(),
out.max_nuclide
.map(nucleide_nuclei::dialects::serpent)
.into_pyobject(py)
.unwrap()
.unbind()
.into_any(),
);
Ok(d)
}
#[pyfunction]
fn origen_tape6_find(py: Python<'_>, text: &str, nuclide: &str) -> PyResult<Option<Py<PyAny>>> {
use pyo3::types::PyDict;
let owned = text.to_owned();
let query = nuclide.to_owned();
let found = py
.detach(move || nucleide_origen_io::Tape6::parse(&owned).map(|t| t.find(&query).cloned()))
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(found.map(|r| {
let d = PyDict::new(py);
d.set_item("nuclide", &r.nuclide).ok();
d.set_item("grams", r.grams).ok();
d.set_item("activity_bq", r.activity_bq).ok();
d.into_any().unbind()
}))
}
#[pyfunction]
fn origen_tape6_total_activity(text: &str) -> PyResult<f64> {
nucleide_origen_io::Tape6::parse(text)
.map(|t| t.total_activity())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn origen_tape9_find(py: Python<'_>, text: &str, nuclide: &str) -> PyResult<Option<Py<PyAny>>> {
use pyo3::types::PyDict;
let owned = text.to_owned();
let query = nuclide.to_owned();
let found = py
.detach(move || {
nucleide_origen_io::Tape9Entry::parse(&owned)
.map(|entries| nucleide_origen_io::Tape9Entry::find(&entries, &query).cloned())
})
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(found.map(|e| {
let d = PyDict::new(py);
d.set_item("nuclide", &e.nuclide).ok();
d.set_item("decay_const", e.decay_const).ok();
d.into_any().unbind()
}))
}
#[pyfunction]
#[pyo3(signature = (text, kind="rtflux"))]
fn cccc_rtflux_npoints(text: &str, kind: &str) -> PyResult<usize> {
let flux_kind = parse_flux_kind(kind)?;
nucleide_cccc_io::FluxFile::parse(flux_kind, text)
.map(|f| f.npoints())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (text, kind="rtflux", index=0))]
fn cccc_rtflux_point(text: &str, kind: &str, index: usize) -> PyResult<Option<Vec<f64>>> {
let flux_kind = parse_flux_kind(kind)?;
let flux = nucleide_cccc_io::FluxFile::parse(flux_kind, text)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(flux.point(index).map(<[f64]>::to_vec))
}
#[pyfunction]
#[pyo3(signature = (text, kind="rtflux"))]
fn cccc_rtflux_total(text: &str, kind: &str) -> PyResult<f64> {
let flux_kind = parse_flux_kind(kind)?;
nucleide_cccc_io::FluxFile::parse(flux_kind, text)
.map(|f| f.total())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
fn parse_flux_kind(kind: &str) -> PyResult<nucleide_cccc_io::rtflux::FluxKind> {
match kind.to_ascii_lowercase().as_str() {
"rtflux" => Ok(nucleide_cccc_io::rtflux::FluxKind::Rtflux),
"atflux" => Ok(nucleide_cccc_io::rtflux::FluxKind::Atflux),
"rzflux" => Ok(nucleide_cccc_io::rtflux::FluxKind::Rzflux),
other => Err(PyValueError::new_err(format!(
"kind must be rtflux|atflux|rzflux, got `{other}`"
))),
}
}
#[pyfunction]
fn cccc_isotxs_find(py: Python<'_>, text: &str, label: &str) -> PyResult<Option<Py<PyAny>>> {
use pyo3::types::PyDict;
let owned = text.to_owned();
let query = label.to_owned();
let found = py
.detach(move || {
nucleide_cccc_io::IsotxsLib::parse(&owned).map(|lib| lib.find(&query).cloned())
})
.map_err(|e| PyValueError::new_err(e.to_string()))?;
Ok(found.map(|n| {
let d = PyDict::new(py);
d.set_item("label", &n.label).ok();
d.set_item("zaid", &n.zaid).ok();
d.set_item("groups", n.groups).ok();
d.set_item("total_xs", n.total_xs.clone()).ok();
d.into_any().unbind()
}))
}
#[pyfunction]
fn cccc_isotxs_len(text: &str) -> PyResult<usize> {
nucleide_cccc_io::IsotxsLib::parse(text)
.map(|lib| lib.len())
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
fn fispact_is_output(path: &str) -> bool {
nucleide_fispact_io::is_fispact_output(path)
}
#[pyfunction]
fn enrichment_prod_per_feed(x_feed: f64, x_prod: f64, x_tail: f64) -> f64 {
nucleide_enrichment::prod_per_feed(x_feed, x_prod, x_tail)
}
#[pyfunction]
fn enrichment_tail_per_feed(x_feed: f64, x_prod: f64, x_tail: f64) -> f64 {
nucleide_enrichment::tail_per_feed(x_feed, x_prod, x_tail)
}
#[pyfunction]
fn enrichment_tail_per_prod(x_feed: f64, x_prod: f64, x_tail: f64) -> f64 {
nucleide_enrichment::tail_per_prod(x_feed, x_prod, x_tail)
}
#[pyfunction]
fn enrichment_feed_per_prod(x_feed: f64, x_prod: f64, x_tail: f64) -> f64 {
nucleide_enrichment::feed_per_prod(x_feed, x_prod, x_tail)
}
#[pyfunction]
fn enrichment_feed_per_tail(x_feed: f64, x_prod: f64, x_tail: f64) -> f64 {
nucleide_enrichment::feed_per_tail(x_feed, x_prod, x_tail)
}
#[pyfunction]
fn enrichment_prod_per_tail(x_feed: f64, x_prod: f64, x_tail: f64) -> f64 {
nucleide_enrichment::prod_per_tail(x_feed, x_prod, x_tail)
}
#[pyfunction]
#[allow(non_snake_case)]
fn enrichment_alphastar_i(alpha: f64, Mstar: f64, M_i: f64) -> f64 {
nucleide_enrichment::alphastar_i(alpha, Mstar, M_i)
}
#[pyfunction]
fn kinetics_from_ifp(
py: Python<'_>,
betas: Vec<f64>,
lambda_gen: f64,
lambdas: Vec<f64>,
) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let params = nucleide_kinetics::KineticParams::from_ifp(betas, lambda_gen, lambdas)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let d = PyDict::new(py);
d.set_item("betas", params.betas()).ok();
d.set_item("lambdas", params.lambdas()).ok();
d.set_item("lambda_gen", params.lambda_gen()).ok();
d.set_item("beta_total", params.beta_total()).ok();
d.set_item("groups", params.groups()).ok();
Ok(d.into_any().unbind())
}
#[pyfunction]
#[pyo3(signature = (tally, selection="total", tolerance=0.5, null_value=0.0))]
fn magic_with(
tally: &PyMeshTally,
selection: &str,
tolerance: f64,
null_value: f64,
) -> PyResult<PyMagicOutput> {
let sel = match selection.trim().to_ascii_lowercase().as_str() {
"total" => nucleide_vr_tools::magic::MagicSelection::Total,
"per_group" | "pergroup" | "per-group" => {
nucleide_vr_tools::magic::MagicSelection::PerGroup
}
other => {
return Err(PyValueError::new_err(format!(
"selection must be total|per_group, got `{other}`"
)));
}
};
let params = nucleide_vr_tools::magic::MagicParams {
tolerance,
null_value,
};
nucleide_vr_tools::magic::magic_with(&tally.inner, sel, params)
.map(|inner| PyMagicOutput { inner })
.map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pyfunction]
#[pyo3(signature = (tally, output, mesh_id=1, window_id=1, upper_bound_ratio=5.0, survival_ratio=3.0, max_split=10, weight_cutoff=1e-38))]
#[allow(clippy::too_many_arguments)]
fn emit_openmc_weight_windows(
py: Python<'_>,
tally: &PyMeshTally,
output: &PyMagicOutput,
mesh_id: u32,
window_id: u32,
upper_bound_ratio: f64,
survival_ratio: f64,
max_split: u32,
weight_cutoff: f64,
) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let options = nucleide_vr_tools::windows::OpenMcOptions {
mesh_id,
window_id,
upper_bound_ratio,
survival_ratio,
max_split,
weight_cutoff,
};
let out = nucleide_vr_tools::windows::emit_openmc_weight_windows(
&output.inner,
&tally.inner,
&options,
)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let d = PyDict::new(py);
d.set_item("xml", out.xml)?;
d.set_item("notes", out.notes)?;
Ok(d.into_any().unbind())
}
#[pyfunction]
#[pyo3(signature = (tally, output, name="ww1", file="wwindows.wwd"))]
fn emit_serpent_wwin(
py: Python<'_>,
tally: &PyMeshTally,
output: &PyMagicOutput,
name: &str,
file: &str,
) -> PyResult<Py<PyAny>> {
use pyo3::types::PyDict;
let out =
nucleide_vr_tools::windows::emit_serpent_wwin(&output.inner, &tally.inner, name, file)
.map_err(|e| PyValueError::new_err(e.to_string()))?;
let d = PyDict::new(py);
d.set_item("text", out.text)?;
d.set_item("card", out.card)?;
d.set_item("notes", out.notes)?;
Ok(d.into_any().unbind())
}
#[pyfunction]
fn mcpl_statsum_validate(comment: &str) -> PyResult<String> {
nucleide_mcpl_io::statsum_validate(comment)
.map(str::to_string)
.map_err(PyValueError::new_err)
}
#[pyfunction]
fn mcpl_statsum_comment(key: &str, value: f64) -> PyResult<String> {
nucleide_mcpl_io::statsum_comment(key, value).map_err(|e| PyValueError::new_err(e.to_string()))
}
#[pymodule]
fn _internal(m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_function(wrap_pyfunction!(version, m)?)?;
m.add_function(wrap_pyfunction!(from_zaid, m)?)?;
m.add_function(wrap_pyfunction!(atomic_mass, m)?)?;
m.add_function(wrap_pyfunction!(natural_abundance, m)?)?;
m.add_function(wrap_pyfunction!(rxname_id, m)?)?;
m.add_function(wrap_pyfunction!(rxname_name, m)?)?;
m.add_function(wrap_pyfunction!(rxname_mt, m)?)?;
m.add_function(wrap_pyfunction!(rxname_label, m)?)?;
m.add_function(wrap_pyfunction!(rxname_doc, m)?)?;
m.add_function(wrap_pyfunction!(rxname_reaction, m)?)?;
m.add_function(wrap_pyfunction!(rxname_id_from_nucdelta, m)?)?;
m.add_function(wrap_pyfunction!(rxname_child, m)?)?;
m.add_function(wrap_pyfunction!(rxname_parent, m)?)?;
m.add_function(wrap_pyfunction!(particle_is_valid, m)?)?;
m.add_function(wrap_pyfunction!(particle_is_valid_pdc, m)?)?;
m.add_function(wrap_pyfunction!(particle_is_hydrogen, m)?)?;
m.add_function(wrap_pyfunction!(particle_is_heavy_ion, m)?)?;
m.add_function(wrap_pyfunction!(dose_f1, m)?)?;
m.add_function(wrap_pyfunction!(dose_lung_model, m)?)?;
m.add_function(wrap_pyfunction!(read_xsdir, m)?)?;
m.add_function(wrap_pyfunction!(read_meshtal, m)?)?;
m.add_function(wrap_pyfunction!(read_wwinp, m)?)?;
m.add_function(wrap_pyfunction!(read_mctal, m)?)?;
m.add_function(wrap_pyfunction!(read_ssw, m)?)?;
m.add_function(wrap_pyfunction!(read_ptrac, m)?)?;
m.add_function(wrap_pyfunction!(read_mcpl, m)?)?;
m.add_function(wrap_pyfunction!(write_mcpl, m)?)?;
m.add_function(wrap_pyfunction!(ssw2mcpl, m)?)?;
m.add_function(wrap_pyfunction!(mcpl2ssw, m)?)?;
m.add_function(wrap_pyfunction!(merge_mcpl, m)?)?;
m.add_function(wrap_pyfunction!(extract_mcpl, m)?)?;
m.add_function(wrap_pyfunction!(mcpl_stats, m)?)?;
m.add_function(wrap_pyfunction!(repair_mcpl, m)?)?;
m.add_function(wrap_pyfunction!(read_endl, m)?)?;
m.add_function(wrap_pyfunction!(endl_endftod, m)?)?;
m.add_function(wrap_pyfunction!(combine_ssw_files, m)?)?;
m.add_function(wrap_pyfunction!(read_chain, m)?)?;
m.add_function(wrap_pyfunction!(build_depletion_system, m)?)?;
m.add_function(wrap_pyfunction!(deplete, m)?)?;
m.add_function(wrap_pyfunction!(deplete_series, m)?)?;
m.add_function(wrap_pyfunction!(simple_xs, m)?)?;
m.add_function(wrap_pyfunction!(scattering_length, m)?)?;
m.add_function(wrap_pyfunction!(decay_energy, m)?)?;
m.add_function(wrap_pyfunction!(decay_branches, m)?)?;
m.add_function(wrap_pyfunction!(decay_branch_fraction, m)?)?;
m.add_function(wrap_pyfunction!(fission_yields, m)?)?;
m.add_function(wrap_pyfunction!(fission_yield, m)?)?;
m.add_function(wrap_pyfunction!(normalize_nuclide, m)?)?;
m.add_function(wrap_pyfunction!(decay_heat, m)?)?;
m.add_function(wrap_pyfunction!(dose_factor, m)?)?;
m.add_function(wrap_pyfunction!(dose_per_g, m)?)?;
m.add_function(wrap_pyfunction!(parse_fgr15_table, m)?)?;
m.add_function(wrap_pyfunction!(fgr15_age_index, m)?)?;
m.add_function(wrap_pyfunction!(mix_by_mass, m)?)?;
m.add_function(wrap_pyfunction!(mix_by_volume, m)?)?;
m.add_function(wrap_pyfunction!(specific_activity, m)?)?;
m.add_function(wrap_pyfunction!(materials_doc_to_xml, m)?)?;
m.add_function(wrap_pyfunction!(expand_elements, m)?)?;
m.add_function(wrap_pyfunction!(collapse_elements, m)?)?;
m.add_function(wrap_pyfunction!(read_serpent, m)?)?;
m.add_function(wrap_pyfunction!(read_usrbin, m)?)?;
m.add_function(wrap_pyfunction!(fluka_material_str, m)?)?;
m.add_function(wrap_pyfunction!(fluka_compound_str, m)?)?;
m.add_function(wrap_pyfunction!(fluka_builtin_set, m)?)?;
m.add_function(wrap_pyfunction!(magic, m)?)?;
m.add_function(wrap_pyfunction!(magic_with, m)?)?;
m.add_function(wrap_pyfunction!(emit_openmc_weight_windows, m)?)?;
m.add_function(wrap_pyfunction!(emit_serpent_wwin, m)?)?;
m.add_function(wrap_pyfunction!(write_ssw, m)?)?;
m.add_function(wrap_pyfunction!(mesh_to_geom, m)?)?;
m.add_function(wrap_pyfunction!(half_life, m)?)?;
m.add_function(wrap_pyfunction!(decay_constant, m)?)?;
m.add_function(wrap_pyfunction!(q_value_capture, m)?)?;
m.add_function(wrap_pyfunction!(q_value_alpha, m)?)?;
m.add_function(wrap_pyfunction!(read_inp, m)?)?;
m.add_function(wrap_pyfunction!(from_formula, m)?)?;
m.add_function(wrap_pyfunction!(activity, m)?)?;
m.add_function(wrap_pyfunction!(to_xml, m)?)?;
m.add_function(wrap_pyfunction!(alara_parse_deck, m)?)?;
m.add_function(wrap_pyfunction!(alara_parse_flux, m)?)?;
m.add_function(wrap_pyfunction!(alara_parse_output, m)?)?;
m.add_function(wrap_pyfunction!(alara_expand_schedule, m)?)?;
m.add_function(wrap_pyfunction!(alara_validate_deck, m)?)?;
m.add_function(wrap_pyfunction!(alara_check_block, m)?)?;
m.add_function(wrap_pyfunction!(alara_flux_total, m)?)?;
m.add_function(wrap_pyfunction!(alara_flux_len, m)?)?;
m.add_function(wrap_pyfunction!(alara_output_totals, m)?)?;
m.add_function(wrap_pyfunction!(alara_output_total_activity, m)?)?;
m.add_function(wrap_pyfunction!(alara_photon_total_strength, m)?)?;
m.add_function(wrap_pyfunction!(alara_schedule_total_time, m)?)?;
m.add_function(wrap_pyfunction!(alara_clearance_eu_table, m)?)?;
m.add_function(wrap_pyfunction!(alara_clearance_index, m)?)?;
m.add_function(wrap_pyfunction!(alara_sum_of_fractions, m)?)?;
m.add_function(wrap_pyfunction!(isotxs_parse, m)?)?;
m.add_function(wrap_pyfunction!(rtflux_parse, m)?)?;
m.add_function(wrap_pyfunction!(cccc_rtflux_npoints, m)?)?;
m.add_function(wrap_pyfunction!(cccc_rtflux_point, m)?)?;
m.add_function(wrap_pyfunction!(cccc_rtflux_total, m)?)?;
m.add_function(wrap_pyfunction!(cccc_isotxs_find, m)?)?;
m.add_function(wrap_pyfunction!(cccc_isotxs_len, m)?)?;
m.add_function(wrap_pyfunction!(partisn_render, m)?)?;
m.add_function(wrap_pyfunction!(partisn_validate, m)?)?;
m.add_function(wrap_pyfunction!(fispact_parse_output, m)?)?;
m.add_function(wrap_pyfunction!(fispact_parse_clearance, m)?)?;
m.add_function(wrap_pyfunction!(fispact_is_output, m)?)?;
m.add_function(wrap_pyfunction!(origen_parse_tape5, m)?)?;
m.add_function(wrap_pyfunction!(origen_parse_tape6, m)?)?;
m.add_function(wrap_pyfunction!(origen_parse_tape9, m)?)?;
m.add_function(wrap_pyfunction!(origen_tape6_find, m)?)?;
m.add_function(wrap_pyfunction!(origen_tape6_total_activity, m)?)?;
m.add_function(wrap_pyfunction!(origen_tape9_find, m)?)?;
m.add_function(wrap_pyfunction!(r2s_from_deck, m)?)?;
m.add_function(wrap_pyfunction!(r2s_from_snapshot, m)?)?;
m.add_function(wrap_pyfunction!(r2s_validate, m)?)?;
m.add_function(wrap_pyfunction!(r2s_expand, m)?)?;
m.add_function(wrap_pyfunction!(r2s_assemble, m)?)?;
m.add_function(wrap_pyfunction!(r2s_tag_zone_strength, m)?)?;
m.add_function(wrap_pyfunction!(r2s_photon_group_sums, m)?)?;
m.add_function(wrap_pyfunction!(r2s_snapshot_inventory, m)?)?;
m.add_function(wrap_pyfunction!(r2s_expand_sweep, m)?)?;
m.add_function(wrap_pyfunction!(kinetics_solve, m)?)?;
m.add_function(wrap_pyfunction!(kinetics_equilibrium, m)?)?;
m.add_function(wrap_pyfunction!(kinetics_initial_rate, m)?)?;
m.add_function(wrap_pyfunction!(kinetics_inhour_rho, m)?)?;
m.add_function(wrap_pyfunction!(kinetics_stable_period, m)?)?;
m.add_function(wrap_pyfunction!(kinetics_prompt_jump, m)?)?;
m.add_function(wrap_pyfunction!(kinetics_from_ifp, m)?)?;
m.add_function(wrap_pyfunction!(unfold_sandii, m)?)?;
m.add_function(wrap_pyfunction!(unfold_forward_fold, m)?)?;
m.add_function(wrap_pyfunction!(plasma_source_particles, m)?)?;
m.add_function(wrap_pyfunction!(plasma_source_emit_cards, m)?)?;
m.add_function(wrap_pyfunction!(plasma_source_spectrum_moments, m)?)?;
m.add_function(wrap_pyfunction!(plasma_source_reactivity, m)?)?;
m.add_function(wrap_pyfunction!(damage_nrt_dpa, m)?)?;
m.add_function(wrap_pyfunction!(damage_arc_dpa, m)?)?;
m.add_function(wrap_pyfunction!(damage_gas_appm, m)?)?;
m.add_function(wrap_pyfunction!(damage_he_dpa_ratio, m)?)?;
m.add_function(wrap_pyfunction!(damage_lindhard_partition, m)?)?;
m.add_function(wrap_pyfunction!(damage_damage_energy, m)?)?;
m.add_function(wrap_pyfunction!(damage_nrt_displacements, m)?)?;
m.add_function(wrap_pyfunction!(damage_arc_efficiency, m)?)?;
m.add_function(wrap_pyfunction!(damage_arc_displacements, m)?)?;
m.add_function(wrap_pyfunction!(damage_fold_uq, m)?)?;
m.add_function(wrap_pyfunction!(tritium_steady, m)?)?;
m.add_function(wrap_pyfunction!(tritium_transient, m)?)?;
m.add_function(wrap_pyfunction!(tritium_time_lag, m)?)?;
m.add_function(wrap_pyfunction!(tritium_breakthrough, m)?)?;
m.add_function(wrap_pyfunction!(tritium_oriani, m)?)?;
m.add_function(wrap_pyfunction!(tritium_langmuir, m)?)?;
m.add_function(wrap_pyfunction!(tritium_irreversible_fill, m)?)?;
m.add_function(wrap_pyfunction!(tritium_sieverts, m)?)?;
m.add_function(wrap_pyfunction!(tritium_recombination_rate, m)?)?;
m.add_function(wrap_pyfunction!(tritium_layers_steady, m)?)?;
m.add_function(wrap_pyfunction!(tritium_layers_transient, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_rect_smooth, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_five_point_smooth, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_calc_bg, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_gross_count, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_net_counts, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_energy_bins, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_detector_efficiency, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_fit_efficiency, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_xray_lines, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_sdef_decay_source, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_parse_dollar_spe, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_parse_spe, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_read_dollar_spe, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_read_spe, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_parse_lines_tsv, m)?)?;
m.add_function(wrap_pyfunction!(spectroscopy_read_decay_lines, m)?)?;
m.add_function(wrap_pyfunction!(uq_sample_mvn, m)?)?;
m.add_function(wrap_pyfunction!(uq_sample_lhs, m)?)?;
m.add_function(wrap_pyfunction!(uq_sample_lognormal, m)?)?;
m.add_function(wrap_pyfunction!(uq_lognormal_mean, m)?)?;
m.add_function(wrap_pyfunction!(uq_lognormal_cov, m)?)?;
m.add_function(wrap_pyfunction!(uq_sample_mean, m)?)?;
m.add_function(wrap_pyfunction!(uq_sample_cov, m)?)?;
m.add_function(wrap_pyfunction!(uq_check_convergence, m)?)?;
m.add_function(wrap_pyfunction!(uq_perturb_branches, m)?)?;
m.add_function(wrap_pyfunction!(uq_perturb_energies, m)?)?;
m.add_function(wrap_pyfunction!(uq_passthrough, m)?)?;
m.add_function(wrap_pyfunction!(uq_perturb_fission_yields, m)?)?;
m.add_function(wrap_pyfunction!(parse_deck, m)?)?;
m.add_function(wrap_pyfunction!(read_deck, m)?)?;
m.add_function(wrap_pyfunction!(parse_sdef, m)?)?;
m.add_function(wrap_pyfunction!(parse_csg_to_openmc, m)?)?;
m.add_function(wrap_pyfunction!(read_csg_to_openmc, m)?)?;
m.add_function(wrap_pyfunction!(parse_csg_to_serpent, m)?)?;
m.add_function(wrap_pyfunction!(read_csg_to_serpent, m)?)?;
m.add_function(wrap_pyfunction!(parse_csg_to_phits, m)?)?;
m.add_function(wrap_pyfunction!(read_csg_to_phits, m)?)?;
m.add_function(wrap_pyfunction!(parse_csg_to_gdml, m)?)?;
m.add_function(wrap_pyfunction!(read_csg_to_gdml, m)?)?;
m.add_function(wrap_pyfunction!(cumulative_decays, m)?)?;
m.add_function(wrap_pyfunction!(progeny, m)?)?;
m.add_function(wrap_pyfunction!(branching_fraction, m)?)?;
m.add_function(wrap_pyfunction!(decay_mode, m)?)?;
m.add_function(wrap_pyfunction!(chain_edges, m)?)?;
m.add_function(wrap_pyfunction!(armi_to_nucid, m)?)?;
m.add_function(wrap_pyfunction!(nucid_to_armi, m)?)?;
m.add_function(wrap_pyfunction!(mcc3_to_nucid, m)?)?;
m.add_function(wrap_pyfunction!(check_labels, m)?)?;
m.add_function(wrap_pyfunction!(audit_material, m)?)?;
m.add_function(wrap_pyfunction!(separate_material, m)?)?;
m.add_function(wrap_pyfunction!(blend_material, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_value_func, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_swu_per_feed, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_swu_per_prod, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_swu_per_tail, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_prod_per_feed, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_tail_per_feed, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_tail_per_prod, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_feed_per_prod, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_feed_per_tail, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_prod_per_tail, m)?)?;
m.add_function(wrap_pyfunction!(enrichment_alphastar_i, m)?)?;
m.add_function(wrap_pyfunction!(mcpl_statsum_validate, m)?)?;
m.add_function(wrap_pyfunction!(mcpl_statsum_comment, m)?)?;
m.add_class::<PyCusum>()?;
m.add_function(wrap_pyfunction!(emit_cards, m)?)?;
m.add_function(wrap_pyfunction!(emit_drift_table, m)?)?;
m.add_function(wrap_pyfunction!(emit_armi_cards, m)?)?;
m.add_function(wrap_pyfunction!(emit_armi_drift_table, m)?)?;
m.add_class::<PyNuclide>()?;
m.add_class::<PyParticle>()?;
m.add_class::<PyXsdir>()?;
m.add_class::<PyXsdirTable>()?;
m.add_class::<PyMeshtal>()?;
m.add_class::<PyMeshTally>()?;
m.add_class::<PyWwinp>()?;
m.add_class::<PyMctal>()?;
m.add_class::<PySurfSrc>()?;
m.add_class::<PyPtracFile>()?;
m.add_class::<PyMcplFile>()?;
m.add_class::<PyEndlLibrary>()?;
m.add_class::<PyChain>()?;
m.add_class::<PyDepletionSystem>()?;
m.add_class::<PyUsrbinTally>()?;
m.add_class::<PyMagicOutput>()?;
m.add_class::<PyAliasTable>()?;
m.add_class::<PyMeshSourceSampler>()?;
m.add_class::<PyKdeSampler>()?;
m.add_class::<PyCascade>()?;
m.add_class::<PyMaterialsCompendium>()?;
m.add_class::<PyDeckProblem>()?;
m.add_class::<PyInventory>()?;
Ok(())
}