|
1 | 1 | import subprocess |
2 | 2 | import warnings |
3 | | -from contextlib import contextmanager |
4 | | -from dataclasses import asdict, dataclass, replace |
| 3 | +from dataclasses import asdict, dataclass |
5 | 4 | from functools import total_ordering |
6 | | -from typing import Any, Generic, Iterator, List, TypeVar |
7 | | - |
8 | | -from beet import Advancement, Context, ItemModifier, ListOption, LootTable, Predicate |
9 | | -from mecha import ( |
10 | | - AbstractNode, |
11 | | - AstJsonObjectEntry, |
12 | | - AstJsonObjectKey, |
13 | | - AstJsonValue, |
14 | | - AstNbtCompound, |
15 | | - DiagnosticCollection, |
16 | | - DiagnosticError, |
17 | | - Mecha, |
18 | | - rule, |
19 | | -) |
| 5 | +from typing import Any, Generic, List, TypeVar |
| 6 | + |
| 7 | +from beet import ListOption |
20 | 8 | from pydantic import RootModel, field_validator |
21 | | -from tokenstream import SourceLocation, set_location |
22 | 9 |
|
23 | 10 | T = TypeVar('T') |
24 | 11 | import csv |
@@ -142,62 +129,6 @@ def validate_root(cls, value: list[T]|dict[str,T]|T) -> list[T]|dict[str,T]: |
142 | 129 | value = [value] |
143 | 130 | return value # type: ignore |
144 | 131 |
|
145 | | -# TODO 1.20.5: might not need this anymore |
146 | | -class InvokeOnJsonNbt: |
147 | | - """Extendable mixin to run MutatingReducer's rules on nbt within advancements, loot_tables ect...""" |
148 | | - def __init__(self, ctx: Context): |
149 | | - self.ctx = ctx |
150 | | - raise RuntimeError("InvokeOnJsonNbt should not be directly instantiated. It is a mixin for MutatingReducers and should be interited instead") |
151 | | - |
152 | | - @contextmanager |
153 | | - def use_diagnostics(self, diagnostics: DiagnosticCollection) -> Iterator[None]: |
154 | | - """Class is mixed into MutatingReducer, who does have this method. Passed here for type completion""" |
155 | | - raise NotImplementedError() |
156 | | - |
157 | | - def invoke(self, node: AbstractNode, *args: Any, **kwargs: Any) -> Any: |
158 | | - """Class is mixed into MutatingReducer, who does have this method. Passed here for type completion""" |
159 | | - raise NotImplementedError() |
160 | | - |
161 | | - |
162 | | - @rule(AstJsonObjectEntry, key=AstJsonObjectKey(value='nbt')) |
163 | | - @rule(AstJsonObjectEntry, key=AstJsonObjectKey(value='tag')) |
164 | | - def process_nbt_in_json(self, node: AstJsonObjectEntry): |
165 | | - mc = self.ctx.inject(Mecha) |
166 | | - if isinstance(mc.database.current, (Advancement, LootTable, ItemModifier, Predicate)): |
167 | | - if isinstance(node.value, AstJsonValue) and isinstance(node.value.value, str) \ |
168 | | - and node.value.value.startswith("{") and node.value.value.endswith("}"): # excludes location check block/fluid tags - easier than making rule that checks for 'set_nbt' functions on the same json level |
169 | | - try: |
170 | | - nbt = mc.parse(node.value.value.replace("\n", "\\\\n"), type=AstNbtCompound) |
171 | | - except DiagnosticError as exc: |
172 | | - # if parsing failed, give pretty traceback |
173 | | - for d in exc.diagnostics.exceptions: |
174 | | - yield set_location(replace(d, file=mc.database.current), node.value) |
175 | | - return replace(node, value="{}") |
176 | | - |
177 | | - ## TEMP - trial on yielding children rather than using invoke |
178 | | - # with self.use_diagnostics(captured_diagnostics:=DiagnosticCollection()): |
179 | | - # nbt = yield nbt # run all rules on child-node |
180 | | - # print(captured_diagnostics.exceptions) |
181 | | - # print(nbt) |
182 | | - # new_node = replace(node, value=AstJsonValue(value=mc.serialize(nbt, type=AstNbtCompound))) |
183 | | - |
184 | | - with self.use_diagnostics(captured_diagnostics:=DiagnosticCollection()): |
185 | | - processed_nbt = mc.serialize(self.invoke(nbt, type=AstNbtCompound)) |
186 | | - for exc in captured_diagnostics.exceptions: |
187 | | - yield propagate_location(exc, node.value) # set error location to nbt key-value that caused the problem and pass diagnostic back to mecha |
188 | | - |
189 | | - new_node = replace(node, value=AstJsonValue(value=processed_nbt)) |
190 | | - if new_node != node: |
191 | | - return new_node |
192 | | - |
193 | | - return node |
194 | | - |
195 | | -def propagate_location(obj: T, parent_location_obj: Any) -> T: |
196 | | - """a set_location like function propagating diagnostic information for manually invoked rules""" |
197 | | - return set_location(obj, |
198 | | - SourceLocation(pos=parent_location_obj.location.pos+obj.location.pos, lineno=parent_location_obj.location.lineno, colno=parent_location_obj.location.colno+obj.location.colno), # type: ignore |
199 | | - SourceLocation(pos=parent_location_obj.location.pos+obj.end_location.pos, lineno=parent_location_obj.location.lineno, colno=parent_location_obj.location.colno+obj.end_location.colno) # type: ignore |
200 | | - ) |
201 | 132 |
|
202 | 133 | # CSV READING UTILS |
203 | 134 | class CSVCell(str): |
|
0 commit comments