Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
species_retained = []
for matrix in matrices:
species_retained += trim_pfa(matrix, solution.species_names, species_targets, threshold)
# want to ensure retained species are the set of those reachable for each state
species_retained = list(set(species_retained))
if previous_model and len(species_retained) == previous_model.model.n_species:
return previous_model
species_removed = [sp for sp in solution.species_names
if sp not in (species_retained + species_safe)
]
# Cut the exclusion list from the model.
reduced_model = trim(model_file, species_removed, f'reduced_{model_file}')
reduced_model_filename = soln2cti.write(
reduced_model, f'reduced_{reduced_model.n_species}.cti', path=path
)
reduced_model_metrics = sample_metrics(
reduced_model_filename, ignition_conditions, num_threads=num_threads, path=path
)
error = calculate_error(sampled_metrics, reduced_model_metrics)
# If desired, now identify limbo species for future sensitivity analysis
limbo_species = []
if threshold_upper:
species_retained += trim_pfa(
matrix, solution.species_names, species_targets, threshold_upper
)
limbo_species = [
species_retained = []
for matrix in matrices:
species_retained += trim_drg(matrix, solution.species_names, species_targets, threshold)
# want to ensure retained species are the set of those reachable for each state
species_retained = list(set(species_retained))
if previous_model and len(species_retained) == previous_model.model.n_species:
return previous_model
species_removed = [sp for sp in solution.species_names
if sp not in (species_retained + species_safe)
]
# Cut the exclusion list from the model.
reduced_model = trim(model_file, species_removed, f'reduced_{model_file}')
reduced_model_filename = soln2cti.write(
reduced_model, f'reduced_{reduced_model.n_species}.cti', path=path
)
reduced_model_metrics = sample_metrics(
reduced_model_filename, ignition_conditions, num_threads=num_threads, path=path
)
error = calculate_error(sampled_metrics, reduced_model_metrics)
# If desired, now identify limbo species for future sensitivity analysis
limbo_species = []
if threshold_upper:
species_retained += trim_drg(
matrix, solution.species_names, species_targets, threshold_upper
)
limbo_species = [
num_threads : int, optional
Number of CPU threads to use for performing simulations in parallel.
Optional; default = 1, in which the multiprocessing module is not used.
If 0, then use the available number of cores minus one. Otherwise,
use the specified number of threads.
Returns
-------
species_errors : numpy.ndarray
Maximum errors induced by removal of each limbo species
"""
species_errors = np.zeros(len(species_limbo))
with TemporaryDirectory() as temp_dir:
for idx, species in enumerate(species_limbo):
test_model = trim(
starting_model.filename, [species], f'reduced_model_{species}.cti',
phase_name=phase_name
)
test_model_file = soln2cti.write(
test_model, f'reduced_model_{species}.cti', path=temp_dir
)
reduced_model_metrics = sample_metrics(
test_model_file, ignition_conditions, phase_name=phase_name,
num_threads=num_threads
)
species_errors[idx] = calculate_error(metrics, reduced_model_metrics)
return species_errors
Return reduced model and associated metadata
"""
solution = ct.Solution(model_file, phase_name)
species_removed = [sp for sp in solution.species_names
if importance_coeffs[sp] < threshold
and sp not in species_safe
]
if (previous_model and
len(species_removed) == solution.n_species - previous_model.model.n_species
):
return previous_model
# Cut the exclusion list from the model.
reduced_model = trim(
model_file, species_removed, f'reduced_{model_file}', phase_name=phase_name
)
reduced_model_filename = soln2cti.write(
reduced_model, f'reduced_{reduced_model.n_species}.cti', path=path
)
reduced_model_metrics = sample_metrics(
reduced_model_filename, ignition_conditions, phase_name=phase_name,
num_threads=num_threads, path=path
)
error = calculate_error(sampled_metrics, reduced_model_metrics)
return ReducedModel(
model=reduced_model, filename=reduced_model_filename, error=error
)
# this will be the only evaluation.
species_errors = evaluate_species_errors(
current_model, ignition_conditions, initial_metrics, species_limbo,
phase_name=phase_name, num_threads=num_threads
)
# Use a temporary directory to avoid cluttering the working directory with
# all the temporary model files
with TemporaryDirectory() as temp_dir:
while species_limbo:
# use difference between error and current error to find species to remove
idx = np.argmin(np.abs(species_errors - current_model.error))
species_errors = np.delete(species_errors, idx)
species_remove = species_limbo.pop(idx)
test_model = trim(
current_model.filename, [species_remove], f'reduced_model_{species_remove}.cti',
phase_name=phase_name
)
test_model_file = soln2cti.write(
test_model, output_filename=f'reduced_model_{species_remove}.cti', path=temp_dir
)
reduced_model_metrics = sample_metrics(
test_model_file, ignition_conditions, phase_name=phase_name,
num_threads=num_threads, path=path
)
error = calculate_error(initial_metrics, reduced_model_metrics)
logging.info(f'{test_model.n_species:^17} | {species_remove:^17} | {error:^.2f}')
# Ensure new error isn't too high