This repository has been archived by the owner on Aug 13, 2018. It is now read-only.
-
-
Notifications
You must be signed in to change notification settings - Fork 10
/
env.py
260 lines (220 loc) · 8.13 KB
/
env.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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
from __future__ import absolute_import, division, print_function
import json
import os
import sys
import shutil
import requests
import logging
import tempfile
import zipfile
from subprocess import Popen, PIPE
import warnings
from .exceptions import CondaException
from .utils import shell_out
mini_file = "Miniconda-latest.sh"
miniconda_urls = {"linux": "https://repo.continuum.io/miniconda/"
"Miniconda3-latest-Linux-x86_64.sh",
"darwin": "https://repo.continuum.io/miniconda/"
"Miniconda3-latest-MacOSX-x86_64.sh",
"win": "https://repo.continuum.io/miniconda/"
"Miniconda3-latest-Windows-x86_64.exe"}
logger = logging.getLogger(__name__)
here = os.path.dirname(__file__)
def miniconda_url():
"""What to download for this platform"""
if sys.platform.startswith('linux'):
url = miniconda_urls['linux']
elif sys.platform.startswith('darwin'): # pragma: no cover
url = miniconda_urls['darwin']
else: # pragma: no cover
url = miniconda_urls['win']
if not sys.maxsize > 2 ** 32: # pragma: no cover
# 64bit check
url = url.replace("_64", "")
return url
class CondaCreator(object):
"""
Create Conda Env
The parameters below can generally be guessed from the system.
If `conda info` is required, it will only be run on the first
invocation, and the result cached.
Parameters
----------
conda_root: str
Location of a conda installation. The conda executable is expected
at /bin/conda within.
If None, runs `conda info` to find out relevant information.
If no conda is found for `conda info`, or no conda executable exists
within the given location, will download and install miniconda
at that location. If the value is None, that location is within the
source tree.
conda_envs: str
directory in which to create environments; usually within the
source directory (so as not to pollute normal usage of conda)
miniconda_url: str
location to download miniconda from, if needed. Uses `miniconda_urls`
for the appropriate platform if not given.
channels: list of str
Channels to specify to conda. Note that defaults in .condarc will also
be included
conda_pkgs: str
Directory containing cached conda packages, normally within conda_root.
"""
conda_info = {}
def __init__(self, conda_root=None, conda_envs=None, miniconda_url=None,
channels=None, conda_pkgs=None):
if conda_root is None:
self._get_conda_info()
if self.conda_info:
self.conda_root = self.conda_info['conda_prefix']
else:
self.conda_root = os.path.join(here, 'tmp_conda')
else:
self.conda_root = conda_root
self.conda_bin = os.path.join(self.conda_root, 'bin', 'conda')
if not os.path.exists(self.conda_bin):
self._install_miniconda(self.conda_root, miniconda_url)
self.conda_envs = conda_envs or os.sep.join([here, 'tmp_conda', 'envs'])
self.conda_pkgs = conda_pkgs
self.channels = channels or []
def _install_miniconda(self, root, url):
if os.path.exists(root):
# conda wants to create the dir - plus this errors if location
# is not empty
os.rmdir(root)
url = url or miniconda_url()
tmp = tempfile.mkdtemp()
minifile = os.path.join(tmp, 'Miniconda3')
logger.debug("Downloading latest Miniconda.sh")
r = requests.get(url, stream=True)
with open(minifile, 'wb') as f:
for chunk in r.iter_content(chunk_size=1024):
if chunk:
f.write(chunk)
f.flush()
install_cmd = "bash {0} -b -p {1}".format(minifile, root).split()
logger.debug("Installing Miniconda in {0}".format(root))
proc = Popen(install_cmd, stdout=PIPE, stderr=PIPE)
out, err = proc.communicate()
logger.debug(out)
logger.debug(err)
self.conda_info['conda_prefix'] = root
def _get_conda_info(self):
"""Ask a conda on PATH where it is installed"""
if self.conda_info:
# already did this before
return
try:
self.conda_info.update(json.loads(shell_out(
['conda', 'info', '--json'])))
except (OSError, IOError):
warnings.warn('No conda found on PATH')
def _create_env(self, env_name, packages=None, remove=False):
"""
Create Conda env environment. If env_name is found in self.conda_envs,
if will be used without checking the existence of any given packages
Parameters
----------
env_name : str
packages : list
remove : bool
remove environment should it exist - start from
Returns
-------
path : str
path to newly created conda environment
"""
env_path = os.path.join(self.conda_envs, env_name)
if os.path.exists(env_path):
if not remove:
# assume env is OK, ignore packages.
return env_path
shutil.rmtree(env_path)
if not isinstance(packages, list):
raise TypeError("Packages must be a list of strings")
ch = []
[ch.extend(['-c', c]) for c in self.channels]
cmd = [self.conda_bin, 'create', '-p', env_path, '-y',
'-q'] + packages + ch
logger.info("Creating new env {0}".format(env_name))
logger.info(' '.join(cmd))
env = dict(os.environ)
if self.conda_pkgs:
env['CONDA_PKGS_DIRS'] = self.conda_pkgs
proc = Popen(cmd, stdout=PIPE, stderr=PIPE, env=env)
out, err = proc.communicate()
logger.debug(out)
logger.debug(err)
env_python = os.path.join(env_path, 'bin', 'python')
if not os.path.exists(env_python):
raise CondaException("Failed to create Python binary at %s."
"" % env_python)
return env_path
def find_env(self, env_name):
"""
Find full path to env_name
Parameters
----------
env_name : str
Returns
-------
path : str
path to conda environment
"""
env_path = os.path.join(self.conda_envs, env_name)
if os.path.exists(env_path):
return env_path
def create_env(self, env_name, packages=None, remove=False):
"""
Create zipped directory of a conda environment
Parameters
----------
env_name : str
packages : list
remove : bool
remove environment should it exist
Returns
-------
path : str
path to zipped conda environment
"""
if not packages:
env_path = self.find_env(env_name)
else:
env_path = self._create_env(env_name, packages, remove)
return zip_path(env_path)
def zip_path(path, out_file=None):
"""
Zip up directory
Parameters
----------
path : string
out_path : str
Output zip-file; if note given, same as input path with
.zip appended
Returns
-------
path : string
path to zipped file
"""
fname = os.path.basename(path) + '.zip'
env_dir = os.path.dirname(path)
zFile = out_file or os.path.join(env_dir, fname)
# ZipFile does not have a contextmanager in Python 2.6
f = zipfile.ZipFile(zFile, 'w', allowZip64=True)
try:
logger.info('Creating: %s' % zFile)
for root, dirs, files in os.walk(path):
for file in files:
relfile = os.path.join(
os.path.relpath(root, env_dir), file)
absfile = os.path.join(root, file)
try:
os.stat(absfile)
except OSError:
logger.info('Skipping zip for %s' % absfile)
continue
f.write(absfile, relfile)
return zFile
finally:
f.close()