1024090653

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/dfm/emcee/commit/<a class=hub.com/dfm/emcee/commit/<a class="double-link" href="https://git"><a class=hub.com/dfm/emcee/commit/<a class="double-link" href="https://git"><a class=hub.com/dfm/emcee/commit/<a class="double-link" href="https://git"><a class=hub.com/dfm/emcee/commit/8a8c98dec1f483daf74bb904b38a3852c3d2f9c2">8a8c98dec">&lt;a href=&quot;https://github.com/dfm/emcee/commit/</a><a class="double-link" href="https://github.com/dfm/emcee/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/dfm/emcee/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/dfm/emcee/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/dfm/emcee/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/dfm/emcee/commit/8a8c98dec1f483daf74bb904b38a3852c3d2f9c2">8a8c98dec</a><a href="https://github.com/dfm/emcee/commit/8a8c98dec1f483daf74bb904b38a3852c3d2f9c2">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/dfm/emcee/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/dfm/emcee/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/dfm/emcee/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/dfm/emcee/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/dfm/emcee/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/dfm/emcee/commit/8a8c98dec1f483daf74bb904b38a3852c3d2f9c2&quot;&gt;8a8c98dec&lt;/a&gt;&lt;a href=&quot;https://github.com/dfm/emcee/commit/8a8c98dec1f483daf74bb904b38a3852c3d2f9c2&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/dfm/emcee/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/dfm/emcee/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.... (continued)

Pull Request Pull Request #395: Hdf5 swm rmode

Run Details

317 of 370 branches covered (85.68%)

14 of 23 new or added lines in 1 file covered. (60.87%)

1496 of 1570 relevant lines covered (95.29%)

4.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.3

/src/emcee/backends/hdf.py

# -*- coding: utf-8 -*-

from __future__ import division, print_function

import os
from tempfile import NamedTemporaryFile

import numpy as np

from .. import __version__
from .backend import Backend

__all__ = ["HDFBackend", "TempHDFBackend", "does_hdf5_support_longdouble"]


try:
    import h5py
except ImportError:
    h5py = None


def does_hdf5_support_longdouble():
    if h5py is None:
        return False
    with NamedTemporaryFile(
        prefix="emcee-temporary-hdf5", suffix=".hdf5", delete=False
    ) as f:
        f.close()

        with h5py.File(f.name, "w") as hf:
            g = hf.create_group("group")
            g.create_dataset("data", data=np.ones(1, dtype=np.longdouble))
            if g["data"].dtype != np.longdouble:
                return False
        with h5py.File(f.name, "r") as hf:
            if hf["group"]["data"].dtype != np.longdouble:
                return False
    return True


class HDFBackend(Backend):
    """A backend that stores the chain in an HDF5 file using h5py

    .. note:: You must install `h5py <http://www.h5py.org/>`_ to use this
        backend.

    Args:
        filename (str): The name of the HDF5 file where the chain will be
            saved.
        name (str; optional): The name of the group where the chain will
            be saved.
        read_only (bool; optional): If ``True``, the backend will throw a
            ``RuntimeError`` if the file is opened with write access.
        swmr (bool; optional): Enable the Single Writer Multiple Reader 
            mode for HDF5.

    """

    def __init__(
        self,
        filename,
        name="mcmc",
        read_only=False,
        dtype=None,
        compression=None,
        compression_opts=None,
        swmr=False
    ):
        if h5py is None:
            raise ImportError("you must install 'h5py' to use the HDFBackend")
        self.filename = filename
        self.name = name
        self.read_only = read_only
        self.compression = compression
        self.compression_opts = compression_opts
        if dtype is None:
            self.dtype_set = False
            self.dtype = np.float64
        else:
            self.dtype_set = True
            self.dtype = dtype
        self.swmr = swmr

    @property
    def initialized(self):
        if not os.path.exists(self.filename):
            return False
        try:
            with self.open() as f:
                return self.name in f
        except (OSError, IOError):
            return False

    def open(self, mode="r"):
        if self.read_only and mode != "r":
            raise RuntimeError(
                "The backend has been loaded in read-only "
                "mode. Set `read_only = False` to make "
                "changes."
            )
        libver=None
        swmr=False
        if self.swmr:
            libver='latest'
            if mode == 'r':
                swmr=True
        f = h5py.File(self.filename, mode, libver=libver, swmr=swmr)
        if not self.dtype_set and self.name in f:
            g = f[self.name]
            if "chain" in g:
                self.dtype = g["chain"].dtype
                self.dtype_set = True
        return f

    def reset(self, nwalkers, ndim):
        """Clear the state of the chain and empty the backend

        Args:
            nwakers (int): The size of the ensemble
            ndim (int): The number of dimensions

        """
        with self.open("a") as f:
            if self.name in f:
                del f[self.name]

            if self.swmr:
                # enable automatic chunking
                chunks_acc = True # (100,)
                chunks_chain = True # (1,1,ndim)
                chunks_logprob = True # (1,1)
            else:
                chunks_acc = None
                chunks_chain = None
                chunks_logprob = None
            g = f.create_group(self.name)
            g.attrs["version"] = __version__
            g.attrs["nwalkers"] = nwalkers
            g.attrs["ndim"] = ndim
            g.attrs["has_blobs"] = False
            g.attrs["iteration"] = 0
            g.create_dataset(
                "accepted",
                data=np.zeros(nwalkers),
                compression=self.compression,
                compression_opts=self.compression_opts,
                chunks=chunks_acc
            )
            g.create_dataset(
                "chain",
                (0, nwalkers, ndim),
                maxshape=(None, nwalkers, ndim),
                dtype=self.dtype,
                compression=self.compression,
                compression_opts=self.compression_opts,
                chunks=chunks_chain
            )
            g.create_dataset(
                "log_prob",
                (0, nwalkers),
                maxshape=(None, nwalkers),
                dtype=self.dtype,
                compression=self.compression,
                compression_opts=self.compression_opts,
                chunks=chunks_logprob
            )
            if self.swmr:
                f.swmr_mode=True

    def has_blobs(self):
        with self.open() as f:
            return f[self.name].attrs["has_blobs"]

    def get_value(self, name, flat=False, thin=1, discard=0):
        if not self.initialized:
            raise AttributeError(
                "You must run the sampler with "
                "'store == True' before accessing the "
                "results"
            )
        with self.open() as f:
            g = f[self.name]
            iteration = g.attrs["iteration"]
            if iteration <= 0:
                raise AttributeError(
                    "You must run the sampler with "
                    "'store == True' before accessing the "
                    "results"
                )

            if name == "blobs" and not g.attrs["has_blobs"]:
                return None

            v = g[name][discard + thin - 1 : self.iteration : thin]
            if flat:
                s = list(v.shape[1:])
                s[0] = np.prod(v.shape[:2])
                return v.reshape(s)
            return v

    @property
    def shape(self):
        with self.open() as f:
            g = f[self.name]
            return g.attrs["nwalkers"], g.attrs["ndim"]

    @property
    def iteration(self):
        with self.open() as f:
            return f[self.name].attrs["iteration"]

    @property
    def accepted(self):
        with self.open() as f:
            return f[self.name]["accepted"][...]

    @property
    def random_state(self):
        with self.open() as f:
            elements = [
                v
                for k, v in sorted(f[self.name].attrs.items())
                if k.startswith("random_state_")
            ]
        return elements if len(elements) else None

    def grow(self, ngrow, blobs):
        """Expand the storage space by some number of samples

        Args:
            ngrow (int): The number of steps to grow the chain.
            blobs: The current array of blobs. This is used to compute the
                dtype for the blobs array.

        """
        self._check_blobs(blobs)

        with self.open("a") as f:
            g = f[self.name]
            ntot = g.attrs["iteration"] + ngrow
            g["chain"].resize(ntot, axis=0)
            g["log_prob"].resize(ntot, axis=0)
            if blobs is not None:
                has_blobs = g.attrs["has_blobs"]
                if not has_blobs:
                    nwalkers = g.attrs["nwalkers"]
                    dt = np.dtype((blobs.dtype, blobs.shape[1:]))
                    if f.swmr_mode:
                        # swmr mode does not allow the creation of new datasets
                        f.swmr_mode=False
                    g.create_dataset(
                        "blobs",
                        (ntot, nwalkers),
                        maxshape=(None, nwalkers),
                        dtype=dt,
                        compression=self.compression,
                        compression_opts=self.compression_opts,
                    )
                    if self.swmr:
                        f.swmr_mode=True
                else:
                    g["blobs"].resize(ntot, axis=0)
                    if g["blobs"].dtype.shape != blobs.shape[1:]:
                        raise ValueError(
                            "Existing blobs have shape {} but new blobs "
                            "requested with shape {}".format(
                                g["blobs"].dtype.shape, blobs.shape[1:]
                            )
                        )
                g.attrs["has_blobs"] = True
            f.flush() # Notify the reader process that new data has been written

    def save_step(self, state, accepted):
        """Save a step to the backend

        Args:
            state (State): The :class:`State` of the ensemble.
            accepted (ndarray): An array of boolean flags indicating whether
                or not the proposal for each walker was accepted.

        """
        self._check(state, accepted)

        with self.open("a") as f:
            g = f[self.name]
            iteration = g.attrs["iteration"]

            g["chain"][iteration, :, :] = state.coords
            g["log_prob"][iteration, :] = state.log_prob
            if state.blobs is not None:
                g["blobs"][iteration, :] = state.blobs
            g["accepted"][:] += accepted

            for i, v in enumerate(state.random_state):
                g.attrs["random_state_{0}".format(i)] = v

            g.attrs["iteration"] = iteration + 1
            f.flush()


class TempHDFBackend(object):
    def __init__(self, dtype=None, compression=None, compression_opts=None):
        self.dtype = dtype
        self.filename = None
        self.compression = compression
        self.compression_opts = compression_opts

    def __enter__(self):
        f = NamedTemporaryFile(
            prefix="emcee-temporary-hdf5", suffix=".hdf5", delete=False
        )
        f.close()
        self.filename = f.name
        return HDFBackend(
            f.name,
            "test",
            dtype=self.dtype,
            compression=self.compression,
            compression_opts=self.compression_opts,
        )

    def __exit__(self, exception_type, exception_value, traceback):
        os.remove(self.filename)

1	# -- coding: utf-8 --
2
3	from __future__ import division, print_function	5✔
4
5	import os	5✔
6	from tempfile import NamedTemporaryFile	5✔
7
8	import numpy as np	5✔
9
10	from .. import __version__	5✔
11	from .backend import Backend	5✔
12
13	__all__ = ["HDFBackend", "TempHDFBackend", "does_hdf5_support_longdouble"]	5✔
14
15
16	try:	5✔
17	import h5py	5✔
18	except ImportError:
19	h5py = None
20
21
22	def does_hdf5_support_longdouble():	5✔
23	if h5py is None:	5!
24	return False	×
25	with NamedTemporaryFile(	5✔
26	prefix="emcee-temporary-hdf5", suffix=".hdf5", delete=False
27	) as f:
28	f.close()	5✔
29
30	with h5py.File(f.name, "w") as hf:	5✔
31	g = hf.create_group("group")	5✔
32	g.create_dataset("data", data=np.ones(1, dtype=np.longdouble))	5✔
33	if g["data"].dtype != np.longdouble:	5!
34	return False	×
35	with h5py.File(f.name, "r") as hf:	5✔
36	if hf["group"]["data"].dtype != np.longdouble:	5!
37	return False	×
38	return True	5✔
39
40
41	class HDFBackend(Backend):	5✔
42	"""A backend that stores the chain in an HDF5 file using h5py
43
44	.. note:: You must install `h5py <http://www.h5py.org/>`_ to use this
45	backend.
46
47	Args:
48	filename (str): The name of the HDF5 file where the chain will be
49	saved.
50	name (str; optional): The name of the group where the chain will
51	be saved.
52	read_only (bool; optional): If ``True``, the backend will throw a
53	``RuntimeError`` if the file is opened with write access.
54	swmr (bool; optional): Enable the Single Writer Multiple Reader
55	mode for HDF5.
56
57	"""
58
59	def __init__(	5✔
60	self,
61	filename,
62	name="mcmc",
63	read_only=False,
64	dtype=None,
65	compression=None,
66	compression_opts=None,
67	swmr=False
68	):
69	if h5py is None:	5✔
70	raise ImportError("you must install 'h5py' to use the HDFBackend")
71	self.filename = filename	5✔
72	self.name = name	5✔
73	self.read_only = read_only	5✔
74	self.compression = compression	5✔
75	self.compression_opts = compression_opts	5✔
76	if dtype is None:	5✔
77	self.dtype_set = False	5✔
78	self.dtype = np.float64	5✔
79	else:
80	self.dtype_set = True	5✔
81	self.dtype = dtype	5✔
82	self.swmr = swmr	5✔
83
84	@property	5✔
85	def initialized(self):	4✔
86	if not os.path.exists(self.filename):	5✔
87	return False	5✔
88	try:	5✔
89	with self.open() as f:	5✔
90	return self.name in f	5✔
91	except (OSError, IOError):	5✔
92	return False	5✔
93
94	def open(self, mode="r"):	5✔
95	if self.read_only and mode != "r":	5✔
96	raise RuntimeError(	5✔
97	"The backend has been loaded in read-only "
98	"mode. Set `read_only = False` to make "
99	"changes."
100	)
101	libver=None	5✔
102	swmr=False	5✔
103	if self.swmr:	5!
NEW 104	libver='latest'	×
NEW 105	if mode == 'r':	×
NEW 106	swmr=True	×
107	f = h5py.File(self.filename, mode, libver=libver, swmr=swmr)	5✔
108	if not self.dtype_set and self.name in f:	5✔
109	g = f[self.name]	5✔
110	if "chain" in g:	5!
111	self.dtype = g["chain"].dtype	5✔
112	self.dtype_set = True	5✔
113	return f	5✔
114
115	def reset(self, nwalkers, ndim):	5✔
116	"""Clear the state of the chain and empty the backend
117
118	Args:
119	nwakers (int): The size of the ensemble
120	ndim (int): The number of dimensions
121
122	"""
123	with self.open("a") as f:	5✔
124	if self.name in f:	5✔
125	del f[self.name]	5✔
126
127	if self.swmr:	5!
128	# enable automatic chunking
NEW 129	chunks_acc = True # (100,)	×
NEW 130	chunks_chain = True # (1,1,ndim)	×
NEW 131	chunks_logprob = True # (1,1)	×
132	else:
133	chunks_acc = None	5✔
134	chunks_chain = None	5✔
135	chunks_logprob = None	5✔
136	g = f.create_group(self.name)	5✔
137	g.attrs["version"] = __version__	5✔
138	g.attrs["nwalkers"] = nwalkers	5✔
139	g.attrs["ndim"] = ndim	5✔
140	g.attrs["has_blobs"] = False	5✔
141	g.attrs["iteration"] = 0	5✔
142	g.create_dataset(	5✔
143	"accepted",
144	data=np.zeros(nwalkers),
145	compression=self.compression,
146	compression_opts=self.compression_opts,
147	chunks=chunks_acc
148	)
149	g.create_dataset(	5✔
150	"chain",
151	(0, nwalkers, ndim),
152	maxshape=(None, nwalkers, ndim),
153	dtype=self.dtype,
154	compression=self.compression,
155	compression_opts=self.compression_opts,
156	chunks=chunks_chain
157	)
158	g.create_dataset(	5✔
159	"log_prob",
160	(0, nwalkers),
161	maxshape=(None, nwalkers),
162	dtype=self.dtype,
163	compression=self.compression,
164	compression_opts=self.compression_opts,
165	chunks=chunks_logprob
166	)
167	if self.swmr:	5!
NEW 168	f.swmr_mode=True	×
169
170	def has_blobs(self):	5✔
171	with self.open() as f:	5✔
172	return f[self.name].attrs["has_blobs"]	5✔
173
174	def get_value(self, name, flat=False, thin=1, discard=0):	5✔
175	if not self.initialized:	5✔
176	raise AttributeError(	5✔
177	"You must run the sampler with "
178	"'store == True' before accessing the "
179	"results"
180	)
181	with self.open() as f:	5✔
182	g = f[self.name]	5✔
183	iteration = g.attrs["iteration"]	5✔
184	if iteration <= 0:	5✔
185	raise AttributeError(	5✔
186	"You must run the sampler with "
187	"'store == True' before accessing the "
188	"results"
189	)
190
191	if name == "blobs" and not g.attrs["has_blobs"]:	5✔
192	return None	5✔
193
194	v = g[name][discard + thin - 1 : self.iteration : thin]	5✔
195	if flat:	5✔
196	s = list(v.shape[1:])	5✔
197	s[0] = np.prod(v.shape[:2])	5✔
198	return v.reshape(s)	5✔
199	return v	5✔
200
201	@property	5✔
202	def shape(self):	4✔
203	with self.open() as f:	5✔
204	g = f[self.name]	5✔
205	return g.attrs["nwalkers"], g.attrs["ndim"]	5✔
206
207	@property	5✔
208	def iteration(self):	4✔
209	with self.open() as f:	5✔
210	return f[self.name].attrs["iteration"]	5✔
211
212	@property	5✔
213	def accepted(self):	4✔
214	with self.open() as f:	5✔
215	return f[self.name]["accepted"][...]	5✔
216
217	@property	5✔
218	def random_state(self):	4✔
219	with self.open() as f:	5✔
220	elements = [	5✔
221	v
222	for k, v in sorted(f[self.name].attrs.items())
223	if k.startswith("random_state_")
224	]
225	return elements if len(elements) else None	5✔
226
227	def grow(self, ngrow, blobs):	5✔
228	"""Expand the storage space by some number of samples
229
230	Args:
231	ngrow (int): The number of steps to grow the chain.
232	blobs: The current array of blobs. This is used to compute the
233	dtype for the blobs array.
234
235	"""
236	self._check_blobs(blobs)	5✔
237
238	with self.open("a") as f:	5✔
239	g = f[self.name]	5✔
240	ntot = g.attrs["iteration"] + ngrow	5✔
241	g["chain"].resize(ntot, axis=0)	5✔
242	g["log_prob"].resize(ntot, axis=0)	5✔
243	if blobs is not None:	5✔
244	has_blobs = g.attrs["has_blobs"]	5✔
245	if not has_blobs:	5✔
246	nwalkers = g.attrs["nwalkers"]	5✔
247	dt = np.dtype((blobs.dtype, blobs.shape[1:]))	5✔
248	if f.swmr_mode:	5!
249	# swmr mode does not allow the creation of new datasets
NEW 250	f.swmr_mode=False	×
251	g.create_dataset(	5✔
252	"blobs",
253	(ntot, nwalkers),
254	maxshape=(None, nwalkers),
255	dtype=dt,
256	compression=self.compression,
257	compression_opts=self.compression_opts,
258	)
259	if self.swmr:	5!
NEW 260	f.swmr_mode=True	×
261	else:
262	g["blobs"].resize(ntot, axis=0)	5✔
263	if g["blobs"].dtype.shape != blobs.shape[1:]:	5✔
264	raise ValueError(	5✔
265	"Existing blobs have shape {} but new blobs "
266	"requested with shape {}".format(
267	g["blobs"].dtype.shape, blobs.shape[1:]
268	)
269	)
270	g.attrs["has_blobs"] = True	5✔
271	f.flush() # Notify the reader process that new data has been written	5✔
272
273	def save_step(self, state, accepted):	5✔
274	"""Save a step to the backend
275
276	Args:
277	state (State): The :class:`State` of the ensemble.
278	accepted (ndarray): An array of boolean flags indicating whether
279	or not the proposal for each walker was accepted.
280
281	"""
282	self._check(state, accepted)	5✔
283
284	with self.open("a") as f:	5✔
285	g = f[self.name]	5✔
286	iteration = g.attrs["iteration"]	5✔
287
288	g["chain"][iteration, :, :] = state.coords	5✔
289	g["log_prob"][iteration, :] = state.log_prob	5✔
290	if state.blobs is not None:	5✔
291	g["blobs"][iteration, :] = state.blobs	5✔
292	g["accepted"][:] += accepted	5✔
293
294	for i, v in enumerate(state.random_state):	5✔
295	g.attrs["random_state_{0}".format(i)] = v	5✔
296
297	g.attrs["iteration"] = iteration + 1	5✔
298	f.flush()	5✔
299
300
301	class TempHDFBackend(object):	5✔
302	def __init__(self, dtype=None, compression=None, compression_opts=None):	5✔
303	self.dtype = dtype	5✔
304	self.filename = None	5✔
305	self.compression = compression	5✔
306	self.compression_opts = compression_opts	5✔
307
308	def __enter__(self):	5✔
309	f = NamedTemporaryFile(	5✔
310	prefix="emcee-temporary-hdf5", suffix=".hdf5", delete=False
311	)
312	f.close()	5✔
313	self.filename = f.name	5✔
314	return HDFBackend(	5✔
315	f.name,
316	"test",
317	dtype=self.dtype,
318	compression=self.compression,
319	compression_opts=self.compression_opts,
320	)
321
322	def __exit__(self, exception_type, exception_value, traceback):	5✔
323	os.remove(self.filename)	5✔

dfm / emcee / 1024090653

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous