-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtask.py
85 lines (68 loc) · 2.01 KB
/
task.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
from dataclasses import dataclass, field
from typing import (
Any,
Dict,
Generic,
Iterator,
List,
Optional,
SupportsIndex,
TypeVar,
overload,
Set,
)
import _pickle as cPickle # type: ignore
import bz2
from synth.specification import TaskSpecification
from synth.syntax.program import Program
from synth.syntax.type_system import Type
T = TypeVar("T", bound=TaskSpecification)
@dataclass
class Task(Generic[T]):
type_request: Type
specification: T
solution: Optional[Program] = field(default=None)
metadata: Dict[str, Any] = field(default_factory=lambda: {})
def __str__(self) -> str:
return "{} ({}, spec={}, {})".format(
self.metadata.get("name", "Task"),
self.solution or "no solution",
self.specification,
self.metadata,
)
@dataclass
class Dataset(Generic[T]):
"""
Represents a list of tasks in a given specification.
"""
tasks: List[Task[T]]
metadata: Dict[str, Any] = field(default_factory=lambda: {})
def __len__(self) -> int:
return len(self.tasks)
def __iter__(self) -> Iterator[Task[T]]:
return self.tasks.__iter__()
@overload
def __getitem__(self, key: SupportsIndex) -> Task[T]:
pass
@overload
def __getitem__(self, key: slice) -> List[Task[T]]:
pass
def __getitem__(self, key: Any) -> Any:
return self.tasks.__getitem__(key)
def type_requests(self) -> Set[Type]:
return set([task.type_request for task in self.tasks])
def save(self, path: str) -> None:
"""
Save this dataset in the specified file.
The dataset file is compressed.
"""
with bz2.BZ2File(path, "w") as fd:
cPickle.dump(self, fd)
@classmethod
def load(cls, path: str) -> "Dataset[T]":
"""
Load the dataset object stored in this file.
"""
with bz2.BZ2File(path, "rb") as fd:
dataset: Dataset = cPickle.load(fd)
return dataset