Coverage for moptipyapps / binpacking2d / packing_statistics.py: 83%
280 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-11 04:40 +0000
1"""An extended end result statistics record to represent packings."""
2import argparse
3import os.path
4from dataclasses import dataclass
5from math import isfinite
6from typing import Any, Callable, Final, Generator, Iterable, Mapping, cast
8from moptipy.evaluation.base import (
9 KEY_N,
10 EvaluationDataElement,
11)
12from moptipy.evaluation.end_results import EndResult
13from moptipy.evaluation.end_statistics import CsvReader as EsCsvReader
14from moptipy.evaluation.end_statistics import CsvWriter as EsCsvWriter
15from moptipy.evaluation.end_statistics import (
16 EndStatistics,
17)
18from moptipy.evaluation.end_statistics import (
19 from_end_results as es_from_end_results,
20)
21from moptipy.utils.strings import (
22 num_to_str,
23)
24from pycommons.ds.immutable_map import immutable_mapping
25from pycommons.ds.sequences import reiterable
26from pycommons.io.console import logger
27from pycommons.io.csv import (
28 SCOPE_SEPARATOR,
29 csv_column,
30 csv_scope,
31 csv_select_scope,
32)
33from pycommons.io.csv import CsvReader as CsvReaderBase
34from pycommons.io.csv import CsvWriter as CsvWriterBase
35from pycommons.io.path import Path, file_path, line_writer
36from pycommons.math.sample_statistics import CsvReader as SsCsvReader
37from pycommons.math.sample_statistics import CsvWriter as SsCsvWriter
38from pycommons.math.sample_statistics import SampleStatistics
39from pycommons.strings.string_conv import str_to_num
40from pycommons.types import check_int_range, type_error
42from moptipyapps.binpacking2d.objectives.bin_count import BIN_COUNT_NAME
43from moptipyapps.binpacking2d.packing_result import (
44 _OBJECTIVE_LOWER,
45 _OBJECTIVE_UPPER,
46 KEY_BIN_HEIGHT,
47 KEY_BIN_WIDTH,
48 KEY_N_DIFFERENT_ITEMS,
49 KEY_N_ITEMS,
50 LOWER_BOUNDS_BIN_COUNT,
51 PackingResult,
52)
53from moptipyapps.binpacking2d.packing_result import from_csv as pr_from_csv
54from moptipyapps.binpacking2d.packing_result import from_logs as pr_from_logs
55from moptipyapps.utils.shared import (
56 moptipyapps_argparser,
57 motipyapps_footer_bottom_comments,
58)
61@dataclass(frozen=True, init=False, order=False, eq=False)
62class PackingStatistics(EvaluationDataElement):
63 """
64 An end statistics record of one run of one algorithm on one problem.
66 This record provides the information of the outcome of one application of
67 one algorithm to one problem instance in an immutable way.
68 """
70 #: the original end statistics record
71 end_statistics: EndStatistics
72 #: the number of items in the instance
73 n_items: int
74 #: the number of different items in the instance
75 n_different_items: int
76 #: the bin width
77 bin_width: int
78 #: the bin height
79 bin_height: int
80 #: the objective values evaluated after the optimization
81 objectives: Mapping[str, SampleStatistics]
82 #: the bounds for the objective values (append ".lowerBound" and
83 #: ".upperBound" to all objective function names)
84 objective_bounds: Mapping[str, int | float]
85 #: the bounds for the minimum number of bins of the instance
86 bin_bounds: Mapping[str, int]
88 def __init__(self,
89 end_statistics: EndStatistics,
90 n_items: int,
91 n_different_items: int,
92 bin_width: int,
93 bin_height: int,
94 objectives: Mapping[str, SampleStatistics],
95 objective_bounds: Mapping[str, int | float],
96 bin_bounds: Mapping[str, int]):
97 """
98 Create a consistent instance of :class:`PackingStatistics`.
100 :param end_statistics: the end statistics
101 :param n_items: the number of items
102 :param n_different_items: the number of different items
103 :param bin_width: the bin width
104 :param bin_height: the bin height
105 :param objectives: the objective values computed after the
106 optimization
107 :param bin_bounds: the different bounds for the number of bins
108 :param objective_bounds: the bounds for the objective functions
109 :raises TypeError: if any parameter has a wrong type
110 :raises ValueError: if the parameter values are inconsistent
111 """
112 super().__init__()
113 if not isinstance(end_statistics, EndStatistics):
114 raise type_error(end_statistics, "end_statistics", EndResult)
115 if end_statistics.best_f != objectives[end_statistics.objective]:
116 raise ValueError(
117 f"end_statistics.best_f={end_statistics.best_f}, but "
118 f"objectives[{end_statistics.objective!r}]="
119 f"{objectives[end_statistics.objective]}.")
120 if not isinstance(objectives, Mapping):
121 raise type_error(objectives, "objectives", Mapping)
122 if not isinstance(objective_bounds, Mapping):
123 raise type_error(objective_bounds, "objective_bounds", Mapping)
124 if not isinstance(bin_bounds, Mapping):
125 raise type_error(bin_bounds, "bin_bounds", Mapping)
126 if len(objective_bounds) != (2 * len(objectives)):
127 raise ValueError(f"it is required that there is a lower and an "
128 f"upper bound for each of the {len(objectives)} "
129 f"functions, but we got {len(objective_bounds)} "
130 f"bounds, objectives={objectives}, "
131 f"objective_bounds={objective_bounds}.")
133 for name, stat in objectives.items():
134 if not isinstance(name, str):
135 raise type_error(
136 name, f"name of evaluation[{name!r}]={stat!r}", str)
137 if not isinstance(stat, int | float | SampleStatistics):
138 raise type_error(
139 stat, f"value of evaluation[{name!r}]={stat!r}",
140 (int, float, SampleStatistics))
141 lll: str = csv_scope(name, _OBJECTIVE_LOWER)
142 lower = objective_bounds[lll]
143 if not isfinite(lower):
144 raise ValueError(f"{lll}=={lower}.")
145 uuu = csv_scope(name, _OBJECTIVE_UPPER)
146 upper = objective_bounds[uuu]
147 for value in (stat.minimum, stat.maximum) \
148 if isinstance(stat, SampleStatistics) else (stat, ):
149 if not isfinite(value):
150 raise ValueError(
151 f"non-finite value of evaluation[{name!r}]={value!r}")
152 if not (lower <= value <= upper):
153 raise ValueError(
154 f"it is required that {lll}<=f<={uuu}, but got "
155 f"{lower}, {value}, and {upper}.")
156 bins: Final[SampleStatistics | None] = cast(
157 "SampleStatistics", objectives[BIN_COUNT_NAME]) \
158 if BIN_COUNT_NAME in objectives else None
159 for name2, value2 in bin_bounds.items():
160 if not isinstance(name2, str):
161 raise type_error(
162 name2, f"name of bounds[{name2!r}]={value2!r}", str)
163 check_int_range(value2, f"bounds[{name2!r}]", 1, 1_000_000_000)
164 if (bins is not None) and (bins.minimum < value2):
165 raise ValueError(
166 f"number of bins={bins} is inconsistent with "
167 f"bound {name2!r}={value2}.")
169 object.__setattr__(self, "end_statistics", end_statistics)
170 object.__setattr__(self, "objectives", immutable_mapping(objectives))
171 object.__setattr__(self, "objective_bounds",
172 immutable_mapping(objective_bounds))
173 object.__setattr__(self, "bin_bounds", immutable_mapping(bin_bounds))
174 object.__setattr__(self, "n_different_items", check_int_range(
175 n_different_items, "n_different_items", 1, 1_000_000_000_000))
176 object.__setattr__(self, "n_items", check_int_range(
177 n_items, "n_items", n_different_items, 1_000_000_000_000))
178 object.__setattr__(self, "bin_width", check_int_range(
179 bin_width, "bin_width", 1, 1_000_000_000_000))
180 object.__setattr__(self, "bin_height", check_int_range(
181 bin_height, "bin_height", 1, 1_000_000_000_000))
183 def _tuple(self) -> tuple[Any, ...]:
184 """
185 Create a tuple with all the data of this data class for comparison.
187 :returns: a tuple with all the data of this class, where `None` values
188 are masked out
189 """
190 # noinspection PyProtectedMember
191 return self.end_statistics._tuple()
194def from_packing_results(results: Iterable[PackingResult]) \
195 -> Generator[PackingStatistics, None, None]:
196 """
197 Create packing statistics from a sequence of packing results.
199 :param results: the packing results
200 :returns: a sequence of packing statistics
201 """
202 if not isinstance(results, Iterable):
203 raise type_error(results, "results", Iterable)
204 groups: Final[dict[tuple[str, str, str, str], list[PackingResult]]] \
205 = {}
206 objectives_set: set[str] = set()
207 for i, pr in enumerate(results):
208 if not isinstance(pr, PackingResult):
209 raise type_error(pr, f"end_results[{i}]", PackingResult)
210 setting: tuple[str, str, str, str] = \
211 (pr.end_result.algorithm, pr.end_result.instance,
212 pr.end_result.objective, "" if pr.end_result.encoding is None
213 else pr.end_result.encoding)
214 if setting in groups:
215 groups[setting].append(pr)
216 else:
217 groups[setting] = [pr]
218 objectives_set.update(pr.objectives.keys())
220 if len(groups) <= 0:
221 raise ValueError("results is empty!")
222 if len(objectives_set) <= 0:
223 raise ValueError("results has not objectives!")
224 end_stats: Final[list[EndStatistics]] = []
225 objectives: Final[list[str]] = sorted(objectives_set)
227 for key in sorted(groups.keys()):
228 data = groups[key]
229 pr0 = data[0]
230 n_items: int = pr0.n_items
231 n_different_items: int = pr0.n_different_items
232 bin_width: int = pr0.bin_width
233 bin_height: int = pr0.bin_height
234 used_objective: str = pr0.end_result.objective
235 encoding: str | None = pr0.end_result.encoding
236 if used_objective not in objectives_set:
237 raise ValueError(
238 f"{used_objective!r} not in {objectives_set!r}.")
239 if used_objective != key[2]:
240 raise ValueError(
241 f"used objective={used_objective!r} different "
242 f"from key[2]={key[2]}!?")
243 if (encoding is not None) and (encoding != key[3]):
244 raise ValueError(
245 f"used encoding={encoding!r} different "
246 f"from key[3]={key[3]}!?")
247 objective_bounds: Mapping[str, int | float] = pr0.objective_bounds
248 bin_bounds: Mapping[str, int] = pr0.bin_bounds
249 for i, pr in enumerate(data):
250 if n_items != pr.n_items:
251 raise ValueError(f"n_items={n_items} for data[0] but "
252 f"{pr.n_items} for data[{i}]?")
253 if n_different_items != pr.n_different_items:
254 raise ValueError(
255 f"n_different_items={n_different_items} for data[0] "
256 f"but {pr.n_different_items} for data[{i}]?")
257 if bin_width != pr.bin_width:
258 raise ValueError(
259 f"bin_width={bin_width} for data[0] "
260 f"but {pr.bin_width} for data[{i}]?")
261 if bin_height != pr.bin_height:
262 raise ValueError(
263 f"bin_height={bin_height} for data[0] "
264 f"but {pr.bin_height} for data[{i}]?")
265 if used_objective != pr.end_result.objective:
266 raise ValueError(
267 f"used objective={used_objective!r} for data[0] "
268 f"but {pr.end_result.objective!r} for data[{i}]?")
269 if objective_bounds != pr.objective_bounds:
270 raise ValueError(
271 f"objective_bounds={objective_bounds!r} for data[0] "
272 f"but {pr.objective_bounds!r} for data[{i}]?")
273 if bin_bounds != pr.bin_bounds:
274 raise ValueError(
275 f"bin_bounds={bin_bounds!r} for data[0] "
276 f"but {pr.bin_bounds!r} for data[{i}]?")
278 end_stats.extend(es_from_end_results(pr.end_result for pr in data))
279 if len(end_stats) != 1:
280 raise ValueError(f"got {end_stats} from {data}?")
282 yield PackingStatistics(
283 end_statistics=end_stats[0],
284 n_items=n_items,
285 n_different_items=n_different_items,
286 bin_width=bin_width,
287 bin_height=bin_height,
288 objectives={
289 o: SampleStatistics.from_samples(
290 pr.objectives[o] for pr in data)
291 for o in objectives
292 },
293 objective_bounds=objective_bounds,
294 bin_bounds=bin_bounds,
295 )
296 end_stats.clear()
299def to_csv(results: Iterable[PackingStatistics], file: str) -> Path:
300 """
301 Write a sequence of packing statistics to a file in CSV format.
303 :param results: the end statistics
304 :param file: the path
305 :return: the path of the file that was written
306 """
307 path: Final[Path] = Path(file)
308 logger(f"Writing packing statistics to CSV file {path!r}.")
309 path.ensure_parent_dir_exists()
310 with path.open_for_write() as wt:
311 consumer: Final[Callable[[str], None]] = line_writer(wt)
312 for p in CsvWriter.write(sorted(results)):
313 consumer(p)
314 logger(f"Done writing packing statistics to CSV file {path!r}.")
315 return path
318def from_csv(file: str) -> Iterable[PackingStatistics]:
319 """
320 Load the packing statistics from a CSV file.
322 :param file: the file to read from
323 :returns: the iterable with the packing statistics
324 """
325 path: Final[Path] = file_path(file)
326 logger(f"Now reading CSV file {path!r}.")
327 with path.open_for_read() as rd:
328 yield from CsvReader.read(rd)
329 logger(f"Done reading CSV file {path!r}.")
332class CsvWriter(CsvWriterBase[PackingStatistics]):
333 """A class for CSV writing of :class:`PackingStatistics`."""
335 def __init__(self, data: Iterable[PackingStatistics],
336 scope: str | None = None) -> None:
337 """
338 Initialize the csv writer.
340 :param data: the data to write
341 :param scope: the prefix to be pre-pended to all columns
342 """
343 data = reiterable(data)
344 super().__init__(data, scope)
345 #: the end statistics writer
346 self.__es: Final[EsCsvWriter] = EsCsvWriter((
347 pr.end_statistics for pr in data), scope)
349 bin_bounds_set: Final[set[str]] = set()
350 objectives_set: Final[set[str]] = set()
351 for pr in data:
352 bin_bounds_set.update(pr.bin_bounds.keys())
353 objectives_set.update(pr.objectives.keys())
355 #: the bin bounds
356 self.__bin_bounds: list[str] | None = None \
357 if set.__len__(bin_bounds_set) <= 0 else sorted(bin_bounds_set)
359 #: the objectives
360 objectives: list[SsCsvWriter] | None = None
361 #: the objective names
362 objective_names: tuple[str, ...] | None = None
363 #: the lower bound names
364 objective_lb_names: tuple[str, ...] | None = None
365 #: the upper bound names
366 objective_ub_names: tuple[str, ...] | None = None
367 if set.__len__(objectives_set) > 0:
368 p: Final[str | None] = self.scope
369 objective_names = tuple(sorted(objectives_set))
370 objective_lb_names = tuple(csv_scope(
371 oxx, _OBJECTIVE_LOWER) for oxx in objective_names)
372 objective_ub_names = tuple(csv_scope(
373 oxx, _OBJECTIVE_UPPER) for oxx in objective_names)
374 objectives = [SsCsvWriter(
375 data=(ddd.objectives[k] for ddd in data),
376 scope=csv_scope(p, k), n_not_needed=True, what_short=k,
377 what_long=f"objective function {k}") for k in objective_names]
379 #: the objectives
380 self.__objectives: Final[list[SsCsvWriter] | None] = objectives
381 #: the objective names
382 self.__objective_names: Final[tuple[str, ...] | None] \
383 = objective_names
384 #: the lower bound names
385 self.__objective_lb_names: Final[tuple[str, ...] | None] \
386 = objective_lb_names
387 #: the upper bound names
388 self.__objective_ub_names: Final[tuple[str, ...] | None] \
389 = objective_ub_names
391 def get_column_titles(self) -> Iterable[str]:
392 """Get the column titles."""
393 p: Final[str | None] = self.scope
394 yield from self.__es.get_column_titles()
396 yield csv_scope(p, KEY_BIN_HEIGHT)
397 yield csv_scope(p, KEY_BIN_WIDTH)
398 yield csv_scope(p, KEY_N_ITEMS)
399 yield csv_scope(p, KEY_N_DIFFERENT_ITEMS)
400 if self.__bin_bounds:
401 for b in self.__bin_bounds:
402 yield csv_scope(p, b)
403 if self.__objective_names and self.__objectives:
404 for i, o in enumerate(self.__objectives):
405 yield csv_scope(p, self.__objective_lb_names[i])
406 yield from o.get_column_titles()
407 yield csv_scope(p, self.__objective_ub_names[i])
409 def get_row(self, data: PackingStatistics) -> Iterable[str]:
410 """
411 Render a single packing result record to a CSV row.
413 :param data: the end result record
414 :returns: the iterable with the row text
415 """
416 yield from self.__es.get_row(data.end_statistics)
417 yield repr(data.bin_height)
418 yield repr(data.bin_width)
419 yield repr(data.n_items)
420 yield repr(data.n_different_items)
421 if self.__bin_bounds:
422 for bb in self.__bin_bounds:
423 yield (repr(data.bin_bounds[bb])
424 if bb in data.bin_bounds else "")
425 if self.__objective_names and self.__objectives:
426 lb: Final[tuple[str, ...] | None] = self.__objective_lb_names
427 ub: Final[tuple[str, ...] | None] = self.__objective_ub_names
428 for i, ob in enumerate(self.__objective_names):
429 if lb is not None:
430 ox = lb[i]
431 yield (num_to_str(data.objective_bounds[ox])
432 if ox in data.objective_bounds else "")
433 yield from SsCsvWriter.get_optional_row(
434 self.__objectives[i], data.objectives.get(ob))
435 if ub is not None:
436 ox = ub[i]
437 yield (num_to_str(data.objective_bounds[ox])
438 if ox in data.objective_bounds else "")
440 def get_header_comments(self) -> Iterable[str]:
441 """
442 Get any possible header comments.
444 :returns: the header comments
445 """
446 return ("End Statistics of Bin Packing Experiments",
447 "See the description at the bottom of the file.")
449 def get_footer_comments(self) -> Iterable[str]:
450 """
451 Get any possible footer comments.
453 :returns: the footer comments
454 """
455 yield from self.__es.get_footer_comments()
456 yield ""
457 p: Final[str | None] = self.scope
458 if self.__bin_bounds:
459 for bb in self.__bin_bounds:
460 yield (f"{csv_scope(p, bb)} is a lower "
461 "bound for the number of bins.")
462 if self.__objectives and self.__objective_names:
463 for i, obb in enumerate(self.__objective_names):
464 ob: str = csv_scope(p, obb)
465 ox: str = csv_scope(ob, _OBJECTIVE_LOWER)
466 yield f"{ox}: a lower bound of the {ob} objective function."
467 yield from self.__objectives[i].get_footer_comments()
468 ox = csv_scope(ob, _OBJECTIVE_UPPER)
469 yield f"{ox}: an upper bound of the {ob} objective function."
471 def get_footer_bottom_comments(self) -> Iterable[str]:
472 """Get the bottom footer comments."""
473 yield from motipyapps_footer_bottom_comments(
474 self, "The packing data is assembled using module "
475 "moptipyapps.binpacking2d.packing_statistics.")
476 yield from EsCsvWriter.get_footer_bottom_comments(self.__es)
479class CsvReader(CsvReaderBase[PackingStatistics]):
480 """A class for CSV parsing to get :class:`PackingStatistics`."""
482 def __init__(self, columns: dict[str, int]) -> None:
483 """
484 Create a CSV parser for :class:`EndResult`.
486 :param columns: the columns
487 """
488 super().__init__(columns)
489 #: the end result csv reader
490 self.__es: Final[EsCsvReader] = EsCsvReader(columns)
491 #: the index of the n-items column
492 self.__idx_n_items: Final[int] = csv_column(columns, KEY_N_ITEMS)
493 #: the index of the n different items column
494 self.__idx_n_different: Final[int] = csv_column(
495 columns, KEY_N_DIFFERENT_ITEMS)
496 #: the index of the bin width column
497 self.__idx_bin_width: Final[int] = csv_column(
498 columns, KEY_BIN_WIDTH)
499 #: the index of the bin height column
500 self.__idx_bin_height: Final[int] = csv_column(
501 columns, KEY_BIN_HEIGHT)
502 #: the indices for the objective bounds
503 self.__bin_bounds: Final[tuple[tuple[str, int], ...]] = \
504 csv_select_scope(
505 lambda x: tuple(sorted(((k, v) for k, v in x.items()))),
506 columns, LOWER_BOUNDS_BIN_COUNT)
507 if tuple.__len__(self.__bin_bounds) <= 0:
508 raise ValueError("No bin bounds found?")
509 #: the objective bounds columns
510 self.__objective_bounds: Final[tuple[tuple[str, int], ...]] = \
511 csv_select_scope(
512 lambda x: tuple(sorted(((k, v) for k, v in x.items()))),
513 columns, None,
514 skip_orig_key=lambda s: not str.endswith(
515 s, (_OBJECTIVE_LOWER, _OBJECTIVE_UPPER)))
516 n_bounds: Final[int] = tuple.__len__(self.__objective_bounds)
517 if n_bounds <= 0:
518 raise ValueError("No objective function bounds found?")
519 if (n_bounds & 1) != 0:
520 raise ValueError(f"Number of bounds {n_bounds} should be even.")
521 n_val: Final[tuple[tuple[str, int]]] = ((KEY_N, self.__es.idx_n), )
522 #: the parsers for the per-objective statistics
523 self.__objectives: Final[tuple[tuple[str, SsCsvReader], ...]] = \
524 tuple((ss, csv_select_scope(SsCsvReader, columns, ss, n_val))
525 for ss in sorted({s[0] for s in (str.split(
526 kk[0], SCOPE_SEPARATOR) for kk in
527 self.__objective_bounds) if (list.__len__(s) > 1)
528 and (str.__len__(s[0]) > 0)}))
529 n_objectives: Final[int] = tuple.__len__(self.__objectives)
530 if n_objectives <= 0:
531 raise ValueError("No objectives found?")
532 if (2 * n_objectives) != n_bounds:
533 raise ValueError(
534 f"Number {n_objectives} of objectives "
535 f"inconsistent with number {n_bounds} of bounds.")
537 def parse_row(self, data: list[str]) -> PackingStatistics:
538 """
539 Parse a row of data.
541 :param data: the data row
542 :return: the end result statistics
543 """
544 return PackingStatistics(
545 self.__es.parse_row(data),
546 int(data[self.__idx_n_items]),
547 int(data[self.__idx_n_different]),
548 int(data[self.__idx_bin_width]),
549 int(data[self.__idx_bin_height]),
550 {o: v.parse_row(data) for o, v in self.__objectives},
551 {o: str_to_num(data[v]) for o, v in self.__objective_bounds},
552 {o: int(data[v]) for o, v in self.__bin_bounds},
553 )
556# Run packing-results to stat file if executed as script
557if __name__ == "__main__":
558 parser: Final[argparse.ArgumentParser] = moptipyapps_argparser(
559 __file__, "Build an end-results statistics CSV file.",
560 "This program computes statistics over packing results")
561 def_src: str = "./evaluation/end_results.txt"
562 if not os.path.isfile(def_src):
563 def_src = "./results"
564 parser.add_argument(
565 "source", nargs="?", default=def_src,
566 help="either the directory with moptipy log files or the path to the "
567 "end-results CSV file", type=Path)
568 parser.add_argument(
569 "dest", type=Path, nargs="?",
570 default="./evaluation/end_statistics.txt",
571 help="the path to the end results statistics CSV file to be created")
572 args: Final[argparse.Namespace] = parser.parse_args()
574 src_path: Final[Path] = args.source
575 packing_results: Iterable[PackingResult]
576 if src_path.is_file():
577 logger(f"{src_path!r} identifies as file, load as end-results csv")
578 packing_results = pr_from_csv(src_path)
579 else:
580 logger(f"{src_path!r} identifies as directory, load it as log files")
581 packing_results = pr_from_logs(src_path)
582 to_csv(from_packing_results(results=packing_results), args.dest)