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
130
131
132
|
"""base class for all params"""
import pprint
import simplejson as json
import sys
import os
from ricky.config import PROBABILITIES_DIR, OFFLINE
import ricky.utils as utils
class Params(object):
def __init__(self, *args):
self._api = None
self._url = None
self._params = tuple(args)
def __getitem__(self, name):
"""getter for the param by name"""
for param in self._params:
if param.name == name:
return param
raise ValueError("No param with name %s\n" % name)
def __str__(self):
"""string representation"""
return pprint.pformat(self.as_dict())
def _load_probabilities_json(self, probabilities_file=None):
if probabilities_file:
filepath = probabilities_file
else:
filepath = os.path.join(
PROBABILITIES_DIR,
"%s.json" % (self.api.__class__.__name__)
)
try:
f = open(filepath, 'r')
data = f.read()
f.close()
return json.loads(data)
except json.scanner.JSONDecodeError as e:
sys.stderr.write("Invalid Json - Problem decoding %s\n" % filepath)
sys.stderr.write("%s\n" % e)
sys.exit(1)
except IOError:
sys.stderr.write(
"Could not find probabilities file %s\n" % filepath)
sys.exit(1)
def randomize(
self,
probabilities=None,
probabilities_local=False
):
"""assign random values to all params
if using a probabilities.json file, weight is taken
into account"""
if probabilities:
probabilities_dict = self._load_probabilities_json(probabilities)
elif probabilities_local:
probabilities_dict = self._load_probabilities_json()
else:
probabilities_dict = {}
for param in self._params:
param.randomize(probabilities=probabilities_dict.get(param.name))
@property
def api(self):
"""property setter for im api"""
return self._api
@api.setter
def api(self, cls):
"""property getter for im api"""
self._api = cls
def execute(self):
"""calls the associated api"""
if OFFLINE:
sys.path.append("./photoblaster")
from photoblaster.modules import Pb as _Pb
from photoblaster.config import LOCAL as PBLOCAL
for pbcls in _Pb.__subclasses__():
if pbcls.__name__ == self.__class__.__name__:
params_dict = self.as_dict()
instance = pbcls(**params_dict)
instance.create()
if not PBLOCAL:
instance.file_s3move()
instance.db_send()
return instance.file_dict()
return json.loads(
utils.http_request(self._url, params=self.as_dict())
)
def as_dict(self):
"""displays the params names and values in dictionary form
used by the api call
"""
result = {}
for param in self._params:
result[param.name] = param.value
return result
def as_normalized(self):
return tuple([
{'name': param.name, 'normalized': param.as_normalized()}
for param in self._params
])
def as_serialized(self):
"""
returns params in serialized form to use in a dataset
"""
return tuple([param.as_normalized() for param in self._params])
def from_serialized(self, params_tuple):
"""
sets params from serialized form as acquired from a dataset
"""
for i in range(len(self._params)):
self._params[i].from_normalized(params_tuple[i])
def from_dict(self, params_dict):
"""set param values manually from a dictionary"""
for param in self._params:
if param.name in params_dict.keys():
param.value = params_dict[param.name]
from ricky.params.pbgradient import PbGradient
from ricky.params.pbbreaker import PbBreaker
from ricky.params.pbpattern import PbPattern
from ricky.params.pbgrid import PbGrid
|
