"""
The Pseudo-Boolean (PB) Competition provides benchmark instances in OPB format.
Origin: https://www.cril.univ-artois.fr/PB25/
"""
from __future__ import annotations
import fnmatch
import lzma
import os
import pathlib
import tarfile
import io
from typing import Any, Optional, Callable
from cpmpy.tools.datasets.core import FileDataset
[docs]
class OPBDataset(FileDataset):
"""
Pseudo Boolean Competition (PB) benchmark dataset.
Provides access to benchmark instances from the Pseudo Boolean
competitions. Instances are grouped by `year` and `track` (e.g.,
`"OPT-LIN"`, `"DEC-LIN"`) and stored as `.opb.xz` files.
If the dataset is not available locally, it can be automatically
downloaded and extracted.
- Origin: https://www.cril.univ-artois.fr/PB25/
- Reference: Berre, D. L., Parrain, A. The Pseudo-Boolean Evaluation 2011. JSAT, 7(1), 2012.
To load an instance into a CPMpy model, use :func:`~cpmpy.tools.io.opb.load_opb`.
For examples of using a loader as a dataset ``transform``, see the
:ref:`modeling guide <modeling-datasets>`.
Arguments:
root (str): Root directory where datasets are stored or will be downloaded to (default=".").
year (int): Competition year of the dataset to use (default=2024).
track (str): Track name specifying which subset of the competition instances to load (default="OPT-LIN").
competition (bool): If True, the dataset will filtered on competition-used instances.
transform (callable, optional): Optional transform applied to the instance file path.
target_transform (callable, optional): Optional transform applied to the metadata dictionary.
download (bool): If True, downloads the dataset if it does not exist locally (default=False).
"""
name = "opb"
description = "Pseudo-Boolean Competition benchmark instances."
homepage = "https://www.cril.univ-artois.fr/PB25/"
citation = [
"Berre, D. L., Parrain, A. The Pseudo-Boolean Evaluation 2011. JSAT, 7(1), 2012.",
]
def __init__(
self,
root: str = ".",
year: int = 2024, track: str = "OPT-LIN",
competition: bool = True,
transform: Optional[Callable] = None, target_transform: Optional[Callable] = None,
download: bool = False,
**kwargs: Any
):
"""
Constructor for a dataset object of the PB competition.
Raises:
ValueError: If the dataset directory does not exist and `download=False`,
or if the requested year/track combination is not available.
"""
self.root = pathlib.Path(root)
self.year = year
self.track = track
self.competition = competition
# Check requested dataset
if not str(year).startswith('20'):
raise ValueError("Year must start with '20'")
if not track:
raise ValueError("Track must be specified, e.g. exact-weighted, exact-unweighted, ...")
dataset_dir = self.root / self.name / str(year) / track / ('selected' if self.competition else 'normalized')
super().__init__(
dataset_dir=dataset_dir,
transform=transform, target_transform=target_transform,
download=download, extension=".opb.xz",
**kwargs
)
[docs]
def categories(self) -> dict[str, Any]:
return {
"year": self.year,
"track": self.track
}
[docs]
def download(self):
url = "https://www.cril.univ-artois.fr/"
target = f"PB{str(self.year)[2:]}/benchs/{'normalized' if not self.competition else 'selected'}-PB{str(self.year)[2:]}.tar"
target_download_path = self.root / target
print(f"Downloading OPB {self.year} {self.track} {'competition' if self.competition else 'non-competition'} instances from www.cril.univ-artois.fr")
try:
target_download_path = self._download_file(url, target, destination=str(target_download_path))
except ValueError as e:
raise ValueError(f"No dataset available for year {self.year}. Error: {str(e)}")
# Extract only the specific track folder from the tar
with tarfile.open(target_download_path, "r:*") as tar_ref: # r:* handles .tar, .tar.gz, .tar.bz2, etc.
# Get the main folder name
main_folder = None
for name in tar_ref.getnames():
if "/" in name:
main_folder = name.split("/")[0]
break
if main_folder is None:
raise ValueError("Could not find main folder in tar file")
# Extract only files from the specified track
# Get all unique track names from tar
if not self.competition:
tracks = set()
for member in tar_ref.getmembers():
parts = member.name.split("/")
if len(parts) > 2 and parts[0] == main_folder:
tracks.add(parts[1])
else:
tracks = set()
for member in tar_ref.getmembers():
parts = member.name.split("/")
if len(parts) > 2 and parts[0] == main_folder:
tracks.add(parts[2])
# Check if requested track exists
if self.track not in tracks:
raise ValueError(f"Track '{self.track}' not found in dataset. Available tracks: {sorted(tracks)}")
# Create track folder in root directory
self.dataset_dir.mkdir(parents=True, exist_ok=True)
# Extract files for the specified track
if not self.competition:
prefix = f"{main_folder}/{self.track}/"
else:
prefix = f"{main_folder}/*/{self.track}/"
for member in tar_ref.getmembers():
if fnmatch.fnmatch(member.name, prefix + "*") and member.isfile():
# Path relative to main_folder/track
# Find where the track folder ends and get everything after
track_marker = f"/{self.track}/"
marker_pos = member.name.find(track_marker)
relative_path = member.name[marker_pos + len(track_marker):]
# Flatten: replace "/" with "_" to encode subfolders (some instances have clashing names)
flat_name = relative_path#.replace("/", "_")
target_path = self.dataset_dir / flat_name
os.makedirs(os.path.dirname(target_path), exist_ok=True)
with tar_ref.extractfile(member) as source, open(target_path, "wb") as target:
target.write(source.read())
# Clean up the tar file
target_download_path.unlink()
[docs]
@classmethod
def open(cls, instance: os.PathLike) -> io.TextIOBase:
return lzma.open(instance, 'rt') if str(instance).endswith(".xz") else open(instance)
if __name__ == "__main__":
dataset = OPBDataset(year=2024, track="DEC-LIN", competition=True, download=True)
print("Dataset size:", len(dataset))
print("Instance 0:", dataset[0])