-
Notifications
You must be signed in to change notification settings - Fork 29
/
hyper_phase.py
118 lines (99 loc) · 3.58 KB
/
hyper_phase.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
import copy
import autofit as af
from autofit.tools.phase import Dataset
from autolens.pipeline.phase import abstract
class HyperPhase:
def __init__(self, phase: abstract.AbstractPhase, hyper_name: str):
"""
Abstract HyperPhase. Wraps a phase, performing that phase before performing the action
specified by the run_hyper.
Parameters
----------
phase
A phase
"""
self.phase = phase
self.hyper_name = hyper_name
def run_hyper(self, *args, **kwargs) -> af.Result:
"""
Run the hyper_galaxies phase.
Parameters
----------
args
kwargs
Returns
-------
result
The result of the hyper_galaxies phase.
"""
raise NotImplementedError()
def make_hyper_phase(self) -> abstract.AbstractPhase:
"""
Returns
-------
hyper_phase
A copy of the original phase with a modified name and path
"""
phase = copy.deepcopy(self.phase)
phase.paths.zip()
phase.optimizer = phase.optimizer.copy_with_name_extension(
extension=self.hyper_name + "_" + phase.paths.phase_tag,
remove_phase_tag=True,
)
phase.optimizer.const_efficiency_mode = af.conf.instance.non_linear.get(
"MultiNest", "extension_combined_const_efficiency_mode", bool
)
phase.optimizer.sampling_efficiency = af.conf.instance.non_linear.get(
"MultiNest", "extension_combined_sampling_efficiency", float
)
phase.optimizer.n_live_points = af.conf.instance.non_linear.get(
"MultiNest", "extension_combined_n_live_points", int
)
phase.optimizer.multimodal = af.conf.instance.non_linear.get(
"MultiNest", "extension_combined_multimodal", bool
)
phase.optimizer.evidence_tolerance = af.conf.instance.non_linear.get(
"MultiNest", "extension_combined_evidence_tolerance", float
)
phase.optimizer.terminate_at_acceptance_ratio = af.conf.instance.non_linear.get(
"MultiNest", "extension_combined_terminate_at_acceptance_ratio", bool
)
phase.optimizer.acceptance_ratio_threshold = af.conf.instance.non_linear.get(
"MultiNest", "extension_combined_acceptance_ratio_threshold", float
)
phase.is_hyper_phase = True
phase.customize_priors = self.customize_priors
return phase
def customize_priors(self, results):
pass
def run(
self, dataset: Dataset, results: af.ResultsCollection = None, **kwargs
) -> af.Result:
"""
Run the hyper phase and then the hyper_galaxies phase.
Parameters
----------
dataset
Data
results
Results from previous phases.
kwargs
Returns
-------
result
The result of the phase, with a hyper_galaxies result attached as an attribute with the hyper_name of this
phase.
"""
self.save_dataset(dataset=dataset)
results = (
copy.deepcopy(results) if results is not None else af.ResultsCollection()
)
result = self.phase.run(dataset, results=results, **kwargs)
results.add(self.phase.paths.phase_name, result)
hyper_result = self.run_hyper(
dataset=dataset, results=results, info=info, **kwargs
)
setattr(result, self.hyper_name, hyper_result)
return result
def __getattr__(self, item):
return getattr(self.phase, item)