-
Notifications
You must be signed in to change notification settings - Fork 15
/
minizinc.py
616 lines (521 loc) · 20.8 KB
/
minizinc.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
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
# -*- coding: utf-8 -*-
"""
PyMzn provides functions that mimic and enhance the tools from the libminizinc
library. With these tools, it is possible to compile a MiniZinc model into a
FlatZinc one, solve a given problem and get the output solutions directly into
the python code.
The main function that PyMzn provides is the ``minizinc`` function, which
executes the entire workflow for solving a constranint program encoded in
MiniZinc. Solving a MiniZinc problem with PyMzn is as simple as::
import pymzn
pymzn.minizinc('test.mzn')
The ``minizinc`` function is probably the way to go for most of the problems,
but the ``mzn2fzn`` and ``solns2out`` functions are also included in the public
API to allow for maximum flexibility. The latter two functions are wrappers of
the two homonym MiniZinc tools for, respectively, converting a MiniZinc model
into a FlatZinc one and getting custom output from the solution stream of a
solver.
"""
import os
import logging
import contextlib
from io import BufferedReader, TextIOWrapper
from subprocess import CalledProcessError
from tempfile import NamedTemporaryFile
import pymzn.config as config
from . import solvers
from .solvers import gecode
from .model import MiniZincModel
from ..process import Process
from ..dzn import dict2dzn, dzn2dict
class Solutions:
"""Represents a solution stream from the `minizinc` function.
This class populates lazily but can be referenced and iterated as a list.
Attributes
----------
complete : bool
Whether the stream includes the complete set of solutions. This means
the stream contains all solutions in a satisfiability problem, or it
contains the global optimum for maximization/minimization problems.
"""
# TODO: add option to not save solutions
def __init__(self, stream):
self._stream = stream
self._solns = []
self.complete = False
self._iter = None
self._stats = None
@property
def statistics(self):
self._fetch_all()
return self._stats
def _fetch(self):
try:
solution = next(self._stream)
self._solns.append(solution)
return solution
except StopIteration as stop:
complete, stats = stop.value
self.complete = complete
if stats:
self._stats = stats
self._stream = None
return None
def _fetch_all(self):
while self._stream:
self._fetch()
def __len__(self):
self._fetch_all()
return len(self._solns)
def __next__(self):
if self._stream:
return self._fetch()
else:
if not self._iter:
self._iter = iter(self._solns)
try:
return next(self._iter)
except StopIteration:
self._iter = iter(self._solns)
raise
def __iter__(self):
if not self._stream:
self._iter = iter(self._solns)
return self
def __getitem__(self, key):
self._fetch_all()
return self._solns[key]
def __repr__(self):
self._fetch_all()
return repr(self._solns)
def __str__(self):
self._fetch_all()
return str(self._solns)
def minizinc(
mzn, *dzn_files, data=None, keep=False, include=None, solver=None,
output_mode='dict', output_vars=None, output_dir=None, timeout=None,
all_solutions=False, num_solutions=None, force_flatten=False, args=None,
wait=True, statistics=False, no_output_annotations=False, **kwargs
):
"""Implements the workflow to solve a CSP problem encoded with MiniZinc.
Parameters
----------
mzn : str or MiniZincModel
The minizinc problem to be solved. It can be either a string or an
instance of MiniZincModel. If it is a string, it can be either the path
to the mzn file or the content of the model.
*dzn_files
A list of paths to dzn files to attach to the mzn2fzn execution,
provided as positional arguments; by default no data file is attached.
Data files are meant to be used when there is data that is static across
several minizinc executions.
data : dict
Additional data as a dictionary of variables assignments to supply to
the mzn2fnz function. The dictionary is then automatically converted to
dzn format by the pymzn.dzn function. This property is meant to include
data that dynamically changes across several minizinc executions.
keep : bool
Whether to keep the generated mzn, dzn, fzn and ozn files or not. If
False, the generated files are created as temporary files which will be
deleted right after the problem is solved. Though pymzn generated files
are not originally intended to be kept, this property can be used for
debugging purpose. Notice that in case of error the files are not
deleted even if this parameter is False. Default is False.
include : str or list
One or more additional paths to search for included mzn files.
solver : Solver
An instance of Solver to use to solve the minizinc problem. The default
is pymzn.gecode.
output_mode : 'dzn', 'json', 'item', 'dict'
The desired output format. The default is 'dict' which returns a stream
of solutions decoded as python dictionaries. The 'item' format outputs a
stream of strings as returned by the solns2out tool, formatted according
to the output statement of the MiniZinc model. The 'dzn' and 'json'
formats output a stream of strings formatted in dzn of json
respectively.
output_vars : [str]
A list of output variables. These variables will be the ones included in
the output dictionary. Only available if ouptut_mode='dict'.
output_dir : str
Output directory for files generated by PyMzn. The default (None) is the
temporary directory of your OS (if keep=False) or the current working
directory (if keep=True).
timeout : int
Number of seconds after which the solver should stop the computation and
return the best solution found. This is only available if the solver has
support for a timeout.
all_solutions : bool
Whether all the solutions must be returned. Notice that this can only
be used if the solver supports returning all solutions. Default is False.
num_solutions : int
The upper bound on the number of solutions to be returned. Can only be
used if the solver supports returning a fixed number of solutions.
Default is 1.
force_flatten : bool
Wheter the function should be forced to produce a flat model. Whenever
possible, this function feeds the mzn file to the solver without passing
through the flattener, force_flatten=True prevents this behavior and
always produces a fzn file which is in turn passed to the solver.
args : dict
Arguments for the template engine.
wait : bool
Whether to wait for the solving process to finish before returning the
solution stream.
statistics : bool
Whether to save the statistics of the solver (if supported).
no_output_annotations : bool
Whether to avoid using output annotation when handling output
variables.
**kwargs
Additional arguments to pass to the solver, provided as additional
keyword arguments to this function. Check the solver documentation for
the available arguments.
Returns
-------
Solutions
Returns a list of solutions as a Solutions instance. The actual content
of the stream depends on the output_mode chosen.
"""
if isinstance(mzn, MiniZincModel):
mzn_model = mzn
else:
mzn_model = MiniZincModel(mzn)
if not solver:
solver = config.get('solver', gecode)
elif isinstance(solver, str):
solver = getattr(solvers, solver)
if all_solutions and not solver.support_all:
raise ValueError('The solver cannot return all solutions')
if num_solutions is not None and not solver.support_num:
raise ValueError('The solver cannot return a given number of solutions')
if output_mode != 'dict' and output_vars:
raise ValueError('Output vars only available in `dict` output mode')
if statistics and not solver.support_stats:
raise ValueError('The solver does not support emitting statistics')
if not output_dir:
output_dir = config.get('output_dir', None)
keep = config.get('keep', keep)
if output_mode == 'dict':
if output_vars:
mzn_model.dzn_output(output_vars)
_output_mode = 'item'
else:
_output_mode = 'dzn'
else:
_output_mode = output_mode
no_output_annotations = config.get(
'no_output_annotations', no_output_annotations
)
output_prefix = 'pymzn'
if keep:
mzn_dir = os.getcwd()
if mzn_model.mzn_file:
mzn_dir, mzn_name = os.path.split(mzn_model.mzn_file)
output_prefix, _ = os.path.splitext(mzn_name)
output_dir = output_dir or mzn_dir
output_prefix += '_'
output_file = NamedTemporaryFile(dir=output_dir, prefix=output_prefix,
suffix='.mzn', delete=False, mode='w+',
buffering=1)
mzn_model.compile(
output_file, rewrap=keep, args=args,
no_output_annotations=no_output_annotations
)
output_file.close()
mzn_file = output_file.name
data_file = None
fzn_file = None
ozn_file = None
log = logging.getLogger(__name__)
log.debug('Generated file {}'.format(mzn_file))
force_flatten = (
config.get('force_flatten', force_flatten)
or not solver.support_mzn
or (_output_mode in ['dzn', 'json'] and not solver.support_output_mode)
)
solver_args = {**config.get('solver_args', {}), **kwargs}
stderr = None
try:
if force_flatten:
fzn_file, ozn_file = mzn2fzn(
mzn_file, *dzn_files, data=data, keep_data=keep,
include=include, globals_dir=solver.globals_dir,
output_mode=_output_mode
)
solver_stream, stderr = _solve(
solver, fzn_file, wait=wait, timeout=timeout, output_mode='dzn',
all_solutions=all_solutions, num_solutions=num_solutions,
statistics=statistics, **solver_args
)
out = solns2out(solver_stream, ozn_file)
else:
dzn_files = list(dzn_files)
data, data_file = _prepare_data(mzn_file, data, keep)
if data_file:
dzn_files.append(data_file)
out, stderr = _solve(
solver, mzn_file, *dzn_files, wait=wait, lines=True, data=data,
include=include, timeout=timeout, all_solutions=all_solutions,
num_solutions=num_solutions, output_mode=_output_mode,
statistics=statistics, **solver_args
)
solns = split_solns(out)
if output_mode == 'dict':
solns = _to_dict(solns)
stream = solns
except MiniZincError as err:
err._set(mzn_file, stderr)
raise err
cleanup_files = [] if keep else [mzn_file, data_file, fzn_file, ozn_file]
stream = _cleanup(stream, mzn_file, cleanup_files, stderr)
return Solutions(stream)
def _cleanup(stream, mzn_file, files, stderr=None):
try:
while True:
yield next(stream)
except StopIteration as stop:
log = logging.getLogger(__name__)
with contextlib.suppress(FileNotFoundError):
for _file in files:
if _file:
os.remove(_file)
log.debug('Deleted file: {}'.format(_file))
return stop.value
except MiniZincError as err:
err._set(mzn_file, stderr)
raise err
def _solve(solver, *args, lines=False, wait=False, **kwargs):
if wait:
out, err = solver.solve(*args, **kwargs)
if lines:
return out.splitlines(), err
return out, err
else:
solver_process = solver.solve_start(*args, **kwargs)
if lines:
return solver_process.readlines(), solver_proces.stderr
return solver_process, solver_proces.stderr
def mzn2fzn(mzn_file, *dzn_files, data=None, keep_data=False, globals_dir=None,
include=None, output_mode='item', no_ozn=False):
"""Flatten a MiniZinc model into a FlatZinc one. It executes the mzn2fzn
utility from libminizinc to produce a fzn and ozn files from a mzn one.
Parameters
----------
mzn_file : str
The path to the minizinc problem file.
*dzn_files
A list of paths to dzn files to attach to the mzn2fzn execution,
provided as positional arguments; by default no data file is attached.
data : dict
Additional data as a dictionary of variables assignments to supply to
the mzn2fnz function. The dictionary is then automatically converted to
dzn format by the ``pymzn.dict2dzn`` function. Notice that if the data
provided is too large, a temporary dzn file will be produced.
keep_data : bool
Whether to write the inline data into a dzn file and keep it.
Default is False.
globals_dir : str
The path to the directory for global included files.
include : str or list
One or more additional paths to search for included mzn files when
running ``mzn2fzn``.
output_mode : 'dzn', 'json', 'item'
The desired output format. The default is 'item' which outputs a
stream of strings as returned by the solns2out tool, formatted according
to the output statement of the MiniZinc model. The 'dzn' and 'json'
formats output a stream of strings formatted in dzn of json
respectively.
no_ozn : bool
If True, the ozn file is not produced, False otherwise.
Returns
-------
tuple (str, str)
The paths to the generated fzn and ozn files. If ``no_ozn=True``, the
second argument is None.
"""
args = [config.get('mzn2fzn', 'mzn2fzn')]
if globals_dir:
args += ['-G', globals_dir]
if no_ozn:
args.append('--no-output-ozn')
if output_mode and output_mode in ['dzn', 'json', 'item']:
args += ['--output-mode', output_mode]
if include:
if isinstance(include, str):
include = [include]
elif not isinstance(include, list):
raise TypeError('The path provided is not valid.')
else:
include = []
include += config.get('include', [])
for path in include:
args += ['-I', path]
keep_data = config.get('keep', keep_data)
dzn_files = list(dzn_files)
data, data_file = _prepare_data(mzn_file, data, keep_data)
if data:
args += ['-D', data]
elif data_file:
dzn_files.append(data_file)
args += [mzn_file] + dzn_files
log = logging.getLogger(__name__)
process = None
try:
process = Process(args).run()
except CalledProcessError as err:
log.exception(err.stderr)
raise RuntimeError(err.stderr) from err
if not keep_data:
with contextlib.suppress(FileNotFoundError):
if data_file:
os.remove(data_file)
log.debug('Deleted file: {}'.format(data_file))
mzn_base = os.path.splitext(mzn_file)[0]
fzn_file = '.'.join([mzn_base, 'fzn'])
fzn_file = fzn_file if os.path.isfile(fzn_file) else None
ozn_file = '.'.join([mzn_base, 'ozn'])
ozn_file = ozn_file if os.path.isfile(ozn_file) else None
if fzn_file:
log.debug('Generated file: {}'.format(fzn_file))
if ozn_file:
log.debug('Generated file: {}'.format(ozn_file))
return fzn_file, ozn_file
def _prepare_data(mzn_file, data, keep_data=False):
if not data:
return None, None
if isinstance(data, dict):
data = dict2dzn(data)
elif isinstance(data, str):
data = [data]
elif not isinstance(data, list):
raise TypeError('The additional data provided is not valid.')
log = logging.getLogger(__name__)
if keep_data or sum(map(len, data)) >= config.get('dzn_width', 70):
mzn_base, __ = os.path.splitext(mzn_file)
data_file = mzn_base + '_data.dzn'
with open(data_file, 'w') as f:
f.write('\n'.join(data))
log.debug('Generated file: {}'.format(data_file))
data = None
else:
data = ' '.join(data)
data_file = None
return data, data_file
def _solns2out_process(ozn_file):
args = [config.get('solns2out', 'solns2out'), ozn_file]
process = Process(args)
return process
def solns2out(stream, ozn_file):
"""Wraps the solns2out utility, executes it on the solution stream, and
then returns the output stream.
Parameters
----------
stream : str or BufferedReader
The solution stream returned by the solver.
ozn_file : str
The ozn file path produced by the mzn2fzn function.
Returns
-------
generator of str
The output stream of solns2out encoding the solution stream according to
the provided ozn file.
"""
log = logging.getLogger(__name__)
args = [config.get('solns2out', 'solns2out'), ozn_file]
process = _solns2out_process(ozn_file)
try:
if isinstance(stream, (BufferedReader, TextIOWrapper)):
process.start(stream)
yield from process.readlines()
elif isinstance(stream, Process):
if stream.alive:
process.start(stream.stdout)
yield from process.readlines()
else:
process.run(stream.stdout_data)
yield from process.stdout_data.splitlines()
else:
process.run(stream)
yield from process.stdout_data.splitlines()
except CalledProcessError as err:
log.exception(err.stderr)
raise RuntimeError(err.stderr) from err
SOLN_SEP = '----------'
SEARCH_COMPLETE = '=========='
UNSATISFIABLE = '=====UNSATISFIABLE====='
UNKNOWN = '=====UNKNOWN====='
UNBOUNDED = '=====UNBOUNDED====='
UNSATorUNBOUNDED = '=====UNSATorUNBOUNDED====='
ERROR = '=====ERROR====='
def split_solns(lines):
"""Split the solutions from the output stream of a solver or solns2out"""
_buffer = []
complete = False
for line in lines:
line = line.strip()
if line == SOLN_SEP:
yield '\n'.join(_buffer)
_buffer = []
elif line == SEARCH_COMPLETE:
complete = True
_buffer = []
elif line == UNKNOWN:
raise MiniZincUnknownError
elif line == UNSATISFIABLE:
raise MiniZincUnsatisfiableError
elif line == UNBOUNDED:
raise MiniZincUnboundedError
elif line == UNSATorUNBOUNDED:
raise MiniZincUnsatOrUnboundedError
elif line == ERROR:
raise MiniZincGenericError
else:
_buffer.append(line)
return (complete, '\n'.join(_buffer))
def _to_dict(stream):
try:
while True:
yield dzn2dict(next(stream))
except StopIteration as stop:
return stop.value
class MiniZincError(RuntimeError):
"""Generic error for the MiniZinc functions."""
def __init__(self, msg=None):
super().__init__(msg)
self._mzn_file = None
self._stderr = None
@property
def stderr(self):
return self._stderr
@property
def mzn_file(self):
"""str: the mzn file that generated the error."""
return self._mzn_file
def _set(self, _mzn_file, _stderr):
self._mzn_file = _mzn_file
self._stderr = _stderr
self.args = ('{}: {}'.format(
self._mzn_file, self.args[0] + '\n{}'.format(_stderr)
),)
class MiniZincUnsatisfiableError(MiniZincError):
"""Error raised when a minizinc problem is found to be unsatisfiable."""
def __init__(self):
super().__init__('The problem is unsatisfiable.')
class MiniZincUnknownError(MiniZincError):
"""Error raised when minizinc returns no solution (unknown)."""
def __init__(self):
super().__init__('The solution of the problem is unknown.')
class MiniZincUnboundedError(MiniZincError):
"""Error raised when a minizinc problem is found to be unbounded."""
def __init__(self):
super().__init__('The problem is unbounded.')
class MiniZincUnsatOrUnboundedError(MiniZincError):
"""Error raised when a minizinc problem is found to be unsatisfiable or
unbounded.
"""
def __init__(self):
super().__init__('The problem is unsatisfiable or unbounded.')
class MiniZincGenericError(MiniZincError):
"""Error raised when an error occurs but it is none of the above."""
def __init__(self):
super().__init__('The problem raised an error.')