forked from PPPLDeepLearning/plasma-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathaugment.py
More file actions
129 lines (103 loc) · 4.86 KB
/
Copy pathaugment.py
File metadata and controls
129 lines (103 loc) · 4.86 KB
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
119
120
121
122
123
124
125
126
127
128
129
from __future__ import print_function
import os
import time,sys
import abc
import numpy as np
import random
class ByShotAugmentator(object):
def __init__(self,normalizer):
self.normalizer = normalizer
def __str__(self):
s = self.normalizer.__str__()
s += "\n including by shot augmentation"
return s
def apply(self,shot):
'''
The purpose of the method is to apply normalization to a shot and then optionally apply augmentation with a function that is individual to every shot.
Argument list:
- shot: plasma shot. Should contain an augment function
Config parameters list:
- conf['data']['augment_during_training']: boolean flag, yes or no to augment during training
'''
#first just apply normalization as usual.
self.normalizer.apply(shot)
if shot.augmentation_fn is not None:
shot.augmentation_fn(shot)
def set_inference_mode(self,is_inference):
self.normalizer.set_inference_mode(is_inference)
class AbstractAugmentator(object):
def __init__(self,normalizer,is_inference,conf):
self.conf = conf
self.to_augment_str = self.conf['data']['signal_to_augment']
self.normalizer = normalizer
self.is_inference = is_inference
#set whether we are training or testing
def set_inference(self,is_inference):
self.is_inference = is_inference
def __str__(self):
s = self.normalizer.__str__()
s += "\nIs being augmented!".format(self.to_augment_str)
s += "Signal to augmented: {}\n".format(self.to_augment_str)
s += "Is inference: {}\n".format(self.is_inference)
return s
#for compatibility with code that changes the mode of the normalizer
def set_inference_mode(self,is_inference):
self.normalizer.set_inference_mode(is_inference)
@abc.abstractmethod
def apply(self,shot):
pass
@abc.abstractmethod
def augment(self,sig):
pass
class Augmentator(AbstractAugmentator):
def apply(self,shot):
'''
The purpose of the method is to apply normalization to a shot and then optionally apply augmentation.
During inference, a specific signal (one at a time) is augmented based on the string supplied in the config file.
During training, augment a random signal (again, one at a time) or do not augment at all.
It performs calls to: Augmentator.augment(), random.random.choice
Argument list:
- shot: plasma shot
Config parameters list:
- conf['data']['augment_during_training']: boolean flag, yes or no to augment during training
'''
#first just apply normalization as usual.
self.normalizer.apply(shot)
if self.is_inference:
#during inference, augment a specific signal (one at a time)
to_augment_str = self.to_augment_str
else:
#during training augment a random signal, one at a time
if self.conf['data']['augment_during_training']:
to_augment_str = random.choice([x.description for x in shot.signals])
else:
to_augment_str = None
if to_augment_str is not None:
#FIXME might be better to use search. are we always going to augment 1 signal at a time?
for (i,sig) in enumerate(shot.signals):
if sig.description == to_augment_str:
print ('Augmenting {} signal'.format(sig.description))
shot.signals_dict[sig] = self.augment(shot.signals_dict[sig])
def augment(self,signal,strength=10):
'''
The purpose of the method is to modify a signal specified by a configuration parameter or at random according to
a specific mode. Modes include: noise, zeroing and no augmentation.
It performs calls to: numpy random number generator
Argument list:
- signal: signal
- strength: strength of the noise, measured in standard deviations. Integer, default value: 10
Config parameters list:
- conf['data']['augmentation_mode']: categorical config parameter specifying how to augment. Possible values
include "noise", "zero" and "none" (strings)
Returns:
- signal: augmented signal ... numpy array of numeric types?
'''
if self.conf['data']['augmentation_mode'] == "noise":
return np.random.normal(0,strength,signal.shape)
elif self.conf['data']['augmentation_mode'] == "zero":
return signal*0.0 #if "set to zero" augmentation. Can control in conf.
elif self.conf['data']['augmentation_mode'] == "none":
return signal #if no augmentation. Should be the default in conf.
else:
print("Unknown augmentation mode. Exiting")
exit(-1)