batteryrunners.py at 7c7962b ⋅ nipy/nibabel

# emacs: -*- mode: python-mode; py-indent-offset: 4; indent-tabs-mode: nil -*-
# vi: set ft=python sts=4 ts=4 sw=4 et:
### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
#
#   See COPYING file distributed along with the NiBabel package for the
#   copyright and license terms.
#
### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
""" Battery runner classes and Report classes

These classes / objects are for generic checking / fixing batteries

The ``BatteryRunner`` class will run a series of checks on a single
object.

A check is a callable, of signature ``func(obj, fix=False)`` which
returns a tuple ``(obj, Report)`` for ``func(obj, False)`` or
``func(obj, True)``, where the obj may be a modified object, or a
different object, if ``fix==True``.

To run checks only, and return problem report objects:

>>> def chk(obj, fix=False): # minimal check
...     return obj, Report()
>>> btrun = BatteryRunner((chk,))
>>> reports = btrun.check_only('a string')

To run checks and fixes, returning fixed object and problem report
sequence, with possible fix messages:

>>> fixed_obj, report_seq = btrun.check_fix('a string')

Reports are iterable things, where the elements in the iterations are
``Problems``, with attributes ``error``, ``problem_level``,
``problem_msg``, and possibly empty ``fix_msg``.  The ``problem_level``
is an integer, giving the level of problem, from 0 (no problem) to 50
(very bad problem).  The levels follow the log levels from the logging
module (e.g 40 equivalent to "error" level, 50 to "critical").  The
``error`` can be one of ``None`` if no error to suggest, or an Exception
class that the user might consider raising for this sitation.  The
``problem_msg`` and ``fix_msg`` are human readable strings that should
explain what happened.

=======================
 More about ``checks``
=======================

Checks are callables returning objects and reports, like ``chk`` below,
such that::

   obj, report = chk(obj, fix=False)
   obj, report = chk(obj, fix=True)

For example, for the Analyze header, we need to check the datatype::

    def chk_datatype(hdr, fix=True):
        rep = Report(hdr, HeaderDataError)
        code = int(hdr['datatype'])
        try:
            dtype = AnalyzeHeader._data_type_codes.dtype[code]
        except KeyError:
            rep.problem_level = 40
            rep.problem_msg = 'data code not recognized'
        else:
            if dtype.type is np.void:
                rep.problem_level = 40
                rep.problem_msg = 'data code not supported'
            else:
                return hdr, rep
        if fix:
            rep.fix_problem_msg = 'not attempting fix'
        return hdr, rep

or the bitpix::

    def chk_bitpix(hdr, fix=True):
        rep = Report(HeaderDataError)
        code = int(hdr['datatype'])
        try:
            dt = AnalyzeHeader._data_type_codes.dtype[code]
        except KeyError:
            rep.problem_level = 10
            rep.problem_msg = 'no valid datatype to fix bitpix'
            return hdr, rep
        bitpix = dt.itemsize * 8
        if bitpix == hdr['bitpix']:
            return hdr, rep
        rep.problem_level = 10
        rep.problem_msg = 'bitpix does not match datatype')
        if fix:
            hdr['bitpix'] = bitpix # inplace modification
            rep.fix_msg = 'setting bitpix to match datatype'
        return hdr, ret

or the pixdims::

    def chk_pixdims(hdr, fix=True):
        rep = Report(hdr, HeaderDataError)
        if not np.any(hdr['pixdim'][1:4] < 0):
            return hdr, rep
        rep.problem_level = 40
        rep.problem_msg = 'pixdim[1,2,3] should be positive'
        if fix:
            hdr['pixdim'][1:4] = np.abs(hdr['pixdim'][1:4])
            rep.fix_msg = 'setting to abs of pixdim values'
        return hdr, rep

"""


class BatteryRunner(object):
    """ Class to run set of checks """

    def __init__(self, checks):
        """ Initialize instance from sequence of `checks`

        Parameters
        ----------
        checks : sequence
           sequence of checks, where checks are callables matching
           signature ``obj, rep = chk(obj, fix=False)``.  Checks are run
           in the order they are passed.

        Examples
        --------
        >>> def chk(obj, fix=False): # minimal check
        ...     return obj, Report()
        >>> btrun = BatteryRunner((chk,))
        """
        self._checks = checks

    def check_only(self, obj):
        """ Run checks on `obj` returning reports

        Parameters
        ----------
        obj : anything
           object on which to run checks

        Returns
        -------
        reports : sequence
           sequence of report objects reporting on result of running
           checks (without fixes) on `obj`
        """
        reports = []
        for check in self._checks:
            obj, rep = check(obj, False)
            reports.append(rep)
        return reports

    def check_fix(self, obj):
        """ Run checks, with fixes, on `obj` returning `obj`, reports

        Parameters
        ----------
        obj : anything
           object on which to run checks, fixes

        Returns
        -------
        obj : anything
           possibly modified or replaced `obj`, after fixes
        reports : sequence
           sequence of reports on checks, fixes
        """
        reports = []
        for check in self._checks:
            obj, report = check(obj, True)
            reports.append(report)
        return obj, reports

    def __len__(self):
        return len(self._checks)


class Report(object):

    def __init__(self,
                 error=Exception,
                 problem_level=0,
                 problem_msg='',
                 fix_msg=''):
        """ Initialize report with values

        Parameters
        ----------
        error : None or Exception
           Error to raise if raising error for this check.  If None,
           no error can be raised for this check (it was probably
           normal).
        problem_level : int
           level of problem.  From 0 (no problem) to 50 (severe
           problem).  If the report originates from a fix, then this
           is the level of the problem remaining after the fix.
           Default is 0
        problem_msg : string
           String describing problem detected. Default is ''
        fix_msg : string
           String describing any fix applied.  Default is ''.

        Examples
        --------
        >>> rep = Report()
        >>> rep.problem_level
        0
        >>> rep = Report(TypeError, 10)
        >>> rep.problem_level
        10
        """
        self.error = error
        self.problem_level = problem_level
        self.problem_msg = problem_msg
        self.fix_msg = fix_msg

    def __getstate__(self):
        """ State that defines object

        Returns
        -------
        tup : tuple
        """
        return self.error, self.problem_level, self.problem_msg, self.fix_msg

    def __eq__(self, other):
        """ are two BatteryRunner-like objects equal?

        Parameters
        ----------
        other : object
           report-like object to test equality

        Examples
        --------
        >>> rep = Report(problem_level=10)
        >>> rep2 = Report(problem_level=10)
        >>> rep == rep2
        True
        >>> rep3 = Report(problem_level=20)
        >>> rep == rep3
        False
        """
        return self.__getstate__() == other.__getstate__()

    def __ne__(self, other):
        """ are two BatteryRunner-like objects not equal?

        See docstring for __eq__
        """
        return not self == other

    def __str__(self):
        """ Printable string for object """
        return self.__dict__.__str__()

    @property
    def message(self):
        """ formatted message string, including fix message if present
        """
        if self.fix_msg:
            return '; '.join((self.problem_msg, self.fix_msg))
        return self.problem_msg

    def log_raise(self, logger, error_level=40):
        """ Log problem, raise error if problem >= `error_level`

        Parameters
        ----------
        logger : log
           log object, implementing ``log`` method
        error_level : int, optional
           If ``self.problem_level`` >= `error_level`, raise error
        """
        logger.log(self.problem_level, self.message)
        if self.problem_level and self.problem_level >= error_level:
            if self.error:
                raise self.error(self.problem_msg)

    def write_raise(self, stream, error_level=40, log_level=30):
        """ Write report to `stream`

        Parameters
        ----------
        stream : file-like
           implementing ``write`` method
        error_level : int, optional
           level at which to raise error for problem detected in
           ``self``
        log_level : int, optional
           Such that if `log_level` is >= ``self.problem_level`` we
           write the report to `stream`, otherwise we write nothing.
        """
        if self.problem_level >= log_level:
            stream.write(f'Level {self.problem_level}: {self.message}\n')
        if self.problem_level and self.problem_level >= error_level:
            if self.error:
                raise self.error(self.problem_msg)

Read our documentation on viewing source code .

Navigation	Overlay
`t` Navigate files	`h` Toggle hits
`y` Change url to tip of branch	`m` Toggle misses
`b / v` Jump to prev/next hit line	`p` Toggle partial
`z / x` Jump to prev/next missed or partial line	`1..9` Toggle flags
`shift + o` Open current page in GitHub	`a` Toggle all on
`/ or ?` Show keyboard shortcuts dialog	`c` Toggle context lines or commits

1		# emacs: -- mode: python-mode; py-indent-offset: 4; indent-tabs-mode: nil --
2		# vi: set ft=python sts=4 ts=4 sw=4 et:
3		### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
4		#
5		# See COPYING file distributed along with the NiBabel package for the
6		# copyright and license terms.
7		#
8		### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
9	33	""" Battery runner classes and Report classes
10
11		These classes / objects are for generic checking / fixing batteries
12
13		The ``BatteryRunner`` class will run a series of checks on a single
14		object.
15
16		A check is a callable, of signature ``func(obj, fix=False)`` which
17		returns a tuple ``(obj, Report)`` for ``func(obj, False)`` or
18		``func(obj, True)``, where the obj may be a modified object, or a
19		different object, if ``fix==True``.
20
21		To run checks only, and return problem report objects:
22
23		>>> def chk(obj, fix=False): # minimal check
24		... return obj, Report()
25		>>> btrun = BatteryRunner((chk,))
26		>>> reports = btrun.check_only('a string')
27
28		To run checks and fixes, returning fixed object and problem report
29		sequence, with possible fix messages:
30
31		>>> fixed_obj, report_seq = btrun.check_fix('a string')
32
33		Reports are iterable things, where the elements in the iterations are
34		``Problems``, with attributes ``error``, ``problem_level``,
35		``problem_msg``, and possibly empty ``fix_msg``. The ``problem_level``
36		is an integer, giving the level of problem, from 0 (no problem) to 50
37		(very bad problem). The levels follow the log levels from the logging
38		module (e.g 40 equivalent to "error" level, 50 to "critical"). The
39		``error`` can be one of ``None`` if no error to suggest, or an Exception
40		class that the user might consider raising for this sitation. The
41		``problem_msg`` and ``fix_msg`` are human readable strings that should
42		explain what happened.
43
44		=======================
45		More about ``checks``
46		=======================
47
48		Checks are callables returning objects and reports, like ``chk`` below,
49		such that::
50
51		obj, report = chk(obj, fix=False)
52		obj, report = chk(obj, fix=True)
53
54		For example, for the Analyze header, we need to check the datatype::
55
56		def chk_datatype(hdr, fix=True):
57		rep = Report(hdr, HeaderDataError)
58		code = int(hdr['datatype'])
59		try:
60		dtype = AnalyzeHeader._data_type_codes.dtype[code]
61		except KeyError:
62		rep.problem_level = 40
63		rep.problem_msg = 'data code not recognized'
64		else:
65		if dtype.type is np.void:
66		rep.problem_level = 40
67		rep.problem_msg = 'data code not supported'
68		else:
69		return hdr, rep
70		if fix:
71		rep.fix_problem_msg = 'not attempting fix'
72		return hdr, rep
73
74		or the bitpix::
75
76		def chk_bitpix(hdr, fix=True):
77		rep = Report(HeaderDataError)
78		code = int(hdr['datatype'])
79		try:
80		dt = AnalyzeHeader._data_type_codes.dtype[code]
81		except KeyError:
82		rep.problem_level = 10
83		rep.problem_msg = 'no valid datatype to fix bitpix'
84		return hdr, rep
85		bitpix = dt.itemsize * 8
86		if bitpix == hdr['bitpix']:
87		return hdr, rep
88		rep.problem_level = 10
89		rep.problem_msg = 'bitpix does not match datatype')
90		if fix:
91		hdr['bitpix'] = bitpix # inplace modification
92		rep.fix_msg = 'setting bitpix to match datatype'
93		return hdr, ret
94
95		or the pixdims::
96
97		def chk_pixdims(hdr, fix=True):
98		rep = Report(hdr, HeaderDataError)
99		if not np.any(hdr['pixdim'][1:4] < 0):
100		return hdr, rep
101		rep.problem_level = 40
102		rep.problem_msg = 'pixdim[1,2,3] should be positive'
103		if fix:
104		hdr['pixdim'][1:4] = np.abs(hdr['pixdim'][1:4])
105		rep.fix_msg = 'setting to abs of pixdim values'
106		return hdr, rep
107
108		"""
109
110
111	33	class BatteryRunner(object):
112		""" Class to run set of checks """
113
114	33	def __init__(self, checks):
115		""" Initialize instance from sequence of `checks`
116
117		Parameters
118		----------
119		checks : sequence
120		sequence of checks, where checks are callables matching
121		signature ``obj, rep = chk(obj, fix=False)``. Checks are run
122		in the order they are passed.
123
124		Examples
125		--------
126		>>> def chk(obj, fix=False): # minimal check
127		... return obj, Report()
128		>>> btrun = BatteryRunner((chk,))
129		"""
130	33	self._checks = checks
131
132	33	def check_only(self, obj):
133		""" Run checks on `obj` returning reports
134
135		Parameters
136		----------
137		obj : anything
138		object on which to run checks
139
140		Returns
141		-------
142		reports : sequence
143		sequence of report objects reporting on result of running
144		checks (without fixes) on `obj`
145		"""
146	33	reports = []
147	33	for check in self._checks:
148	33	obj, rep = check(obj, False)
149	33	reports.append(rep)
150	33	return reports
151
152	33	def check_fix(self, obj):
153		""" Run checks, with fixes, on `obj` returning `obj`, reports
154
155		Parameters
156		----------
157		obj : anything
158		object on which to run checks, fixes
159
160		Returns
161		-------
162		obj : anything
163		possibly modified or replaced `obj`, after fixes
164		reports : sequence
165		sequence of reports on checks, fixes
166		"""
167	33	reports = []
168	33	for check in self._checks:
169	33	obj, report = check(obj, True)
170	33	reports.append(report)
171	33	return obj, reports
172
173	33	def __len__(self):
174	33	return len(self._checks)
175
176
177	33	class Report(object):
178
179	33	def __init__(self,
180		error=Exception,
181		problem_level=0,
182		problem_msg='',
183		fix_msg=''):
184		""" Initialize report with values
185
186		Parameters
187		----------
188		error : None or Exception
189		Error to raise if raising error for this check. If None,
190		no error can be raised for this check (it was probably
191		normal).
192		problem_level : int
193		level of problem. From 0 (no problem) to 50 (severe
194		problem). If the report originates from a fix, then this
195		is the level of the problem remaining after the fix.
196		Default is 0
197		problem_msg : string
198		String describing problem detected. Default is ''
199		fix_msg : string
200		String describing any fix applied. Default is ''.
201
202		Examples
203		--------
204		>>> rep = Report()
205		>>> rep.problem_level
206		0
207		>>> rep = Report(TypeError, 10)
208		>>> rep.problem_level
209		10
210		"""
211	33	self.error = error
212	33	self.problem_level = problem_level
213	33	self.problem_msg = problem_msg
214	33	self.fix_msg = fix_msg
215
216	33	def __getstate__(self):
217		""" State that defines object
218
219		Returns
220		-------
221		tup : tuple
222		"""
223	33	return self.error, self.problem_level, self.problem_msg, self.fix_msg
224
225	33	def __eq__(self, other):
226		""" are two BatteryRunner-like objects equal?
227
228		Parameters
229		----------
230		other : object
231		report-like object to test equality
232
233		Examples
234		--------
235		>>> rep = Report(problem_level=10)
236		>>> rep2 = Report(problem_level=10)
237		>>> rep == rep2
238		True
239		>>> rep3 = Report(problem_level=20)
240		>>> rep == rep3
241		False
242		"""
243	33	return self.__getstate__() == other.__getstate__()
244
245	33	def __ne__(self, other):
246		""" are two BatteryRunner-like objects not equal?
247
248		See docstring for __eq__
249		"""
250	0	return not self == other
251
252	33	def __str__(self):
253		""" Printable string for object """
254	33	return self.__dict__.__str__()
255
256	33	@property
257	5	def message(self):
258		""" formatted message string, including fix message if present
259		"""
260	33	if self.fix_msg:
261	33	return '; '.join((self.problem_msg, self.fix_msg))
262	33	return self.problem_msg
263
264	33	def log_raise(self, logger, error_level=40):
265		""" Log problem, raise error if problem >= `error_level`
266
267		Parameters
268		----------
269		logger : log
270		log object, implementing ``log`` method
271		error_level : int, optional
272		If ``self.problem_level`` >= `error_level`, raise error
273		"""
274	33	logger.log(self.problem_level, self.message)
275	33	if self.problem_level and self.problem_level >= error_level:
276	33	if self.error:
277	33	raise self.error(self.problem_msg)
278
279	33	def write_raise(self, stream, error_level=40, log_level=30):
280		""" Write report to `stream`
281
282		Parameters
283		----------
284		stream : file-like
285		implementing ``write`` method
286		error_level : int, optional
287		level at which to raise error for problem detected in
288		``self``
289		log_level : int, optional
290		Such that if `log_level` is >= ``self.problem_level`` we
291		write the report to `stream`, otherwise we write nothing.
292		"""
293	33	if self.problem_level >= log_level:
294	33	stream.write(f'Level {self.problem_level}: {self.message}\n')
295	33	if self.problem_level and self.problem_level >= error_level:
296	33	if self.error:
297	33	raise self.error(self.problem_msg)