-
Notifications
You must be signed in to change notification settings - Fork 20
/
config.py
118 lines (93 loc) · 3.39 KB
/
config.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
# -*- coding: utf-8 -*-
from dataclasses import asdict, dataclass
from enum import Enum
from typing import Any, Dict, Generator, List
from ipyflow.slicing.context import SlicingContext, iter_slicing_contexts
class EnumWithDefault(Enum):
@classmethod
def _missing_(cls, value):
return cls(cls.__default__) # type: ignore
class ExecutionMode(EnumWithDefault):
NORMAL = "normal"
REACTIVE = __default__ = "reactive" # type: ignore
class ExecutionSchedule(EnumWithDefault):
LIVENESS_BASED = "liveness_based"
DAG_BASED = __default__ = "dag_based" # type: ignore
HYBRID_DAG_LIVENESS_BASED = "hybrid_dag_liveness_based"
class FlowDirection(EnumWithDefault):
ANY_ORDER = "any_order"
IN_ORDER = __default__ = "in_order" # type: ignore
class Highlights(EnumWithDefault):
ALL = "all"
NONE = "none"
EXECUTED = __default__ = "executed" # type: ignore
REACTIVE = "reactive"
class ReactivityMode(EnumWithDefault):
BATCH = __default__ = "batch" # type: ignore
INCREMENTAL = "incremental"
# TODO: figure out how to represent different versions of
# same interface (e.g. jupyterlab 4.0, notebook v7, etc)
class Interface(EnumWithDefault):
BENTO = "bento" # ~TODO
COLAB = "colab" # TODO
DATABRICKS = "databricks" # TODO
DATALORE = "datalore" # TODO
DEEPNOTE = "deepnote" # TODO
HEX = "hex" # TODO
IPYTHON = "ipython"
JUPYTER = "jupyter"
JUPYTERLAB = "jupyterlab"
NOTEABLE = "noteable" # TODO
VSCODE = "vscode" # TODO
UNKNOWN = __default__ = "unknown" # type: ignore
class ColorScheme(EnumWithDefault):
NORMAL = __default__ = "normal" # type: ignore
CLASSIC = "classic" # type: ignore
class JsonSerializableMixin:
def to_json(self: Any) -> Dict[str, Any]:
json = {}
for key, value in asdict(self).items():
if isinstance(value, Enum):
value = value.value
if not isinstance(value, (bool, float, str)):
value = str(value)
json[key] = value
return json
@dataclass(frozen=True)
class DataflowSettings(JsonSerializableMixin):
test_context: bool
mark_waiting_symbol_usages_unsafe: bool
mark_typecheck_failures_unsafe: bool
mark_phantom_cell_usages_unsafe: bool
@dataclass
class MutableDataflowSettings(JsonSerializableMixin):
dataflow_enabled: bool
trace_messages_enabled: bool
highlights: Highlights
interface: Interface
static_slicing_enabled: bool
dynamic_slicing_enabled: bool
exec_mode: ExecutionMode
exec_schedule: ExecutionSchedule
flow_order: FlowDirection
reactivity_mode: ReactivityMode
push_reactive_updates: bool
push_reactive_updates_to_cousins: bool
pull_reactive_updates: bool
color_scheme: ColorScheme
warn_out_of_order_usages: bool
lint_out_of_order_usages: bool
syntax_transforms_enabled: bool
syntax_transforms_only: bool
max_external_call_depth_for_tracing: int
is_dev_mode: bool
def slicing_contexts(self) -> List[SlicingContext]:
ret: List[SlicingContext] = []
if self.dynamic_slicing_enabled:
ret.append(SlicingContext.DYNAMIC)
if self.static_slicing_enabled:
ret.append(SlicingContext.STATIC)
return ret
def iter_slicing_contexts(self) -> Generator[None, None, None]:
for _ in iter_slicing_contexts(*self.slicing_contexts()):
yield