Coverage for bim2sim/kernel/decision/__init_

1"""Decision system.

3This package contains:

4 - class Decision (and child classes) for representing decisions

5 - class DecisionBunch for handling collections of Decision elements

6 - class DecisionHandler to handle decisions

7 - functions save() and load() to save to file system

8"""

10from importlib.metadata import version

11import enum

12import hashlib

13import json

14import logging

15from collections import Counter

16from typing import Iterable, Callable, List, Dict, Any, Tuple, Union

18import pint

20from bim2sim.elements.mapping.units import ureg

21logger = logging.getLogger(__name__)

24class DecisionException(Exception):

25 """Base Exception for Decisions"""

28class DecisionSkip(DecisionException):

29 """Exception raised on skipping Decision"""

32class DecisionSkipAll(DecisionException):

33 """Exception raised on skipping all Decisions"""

36class DecisionCancel(DecisionException):

37 """Exception raised on canceling Decisions"""

40class PendingDecisionError(DecisionException):

41 """Exception for unsolved Decisions"""

44class Status(enum.Enum):

45 """Enum for status of Decision"""

46 pending = 1 # decision not yet made

47 ok = 2 # decision made

48 skipped = 5 # decision was skipped

49 error = 6 # invalid answer

52def convert_0_to_0_1(data):

53 converted_data = {

54 'version': '0.1',

55 'checksum_ifc': None,

56 'decisions': {decision: {'value': value} for decision, value in data.items()}

57 }

58 return converted_data

61def convert(from_version, to_version, data):

62 """convert stored decisions to new version"""

63 if from_version == '0' and to_version == '0.1':

64 return convert_0_to_0_1(data)

67class Decision:

68 """A question and a value which should be set to answer the question.

70 Args:

71 question: The question asked to the user

72 console_identifier: Additional information to identify related in

73 console

74 validate_func: callable to validate the users input

75 key: key is used by DecisionBunch to create answer dict

76 global_key: unique key to identify decision. Required for saving

77 allow_skip: set to True to allow skipping the decision and user None as

78 value

79 validate_checksum: if provided, loaded decisions are only valid if

80 checksum matches

81 related: iterable of GUIDs this decision is related to (frontend)

82 context: iterable of GUIDs for additional context to this decision

83 (frontend)

84 default: default answer

85 group: group of decisions this decision belongs to

86 representative_global_keys: list of global keys of elements that this

87 decision also has the answer for

89 Example:

90 >>> decision = Decision("How much is the fish?", allow_skip=True)

92 # the following is usually done by child classes. Here it's a hack to make plain Decisions work

93 >>> decision._validate = lambda value: True

95 >>> decision.value # will raise ValueError

96 Traceback (most recent call last):

97 ValueError: Can't get value from invalid decision.

98 >>> decision.value = 10 # ok

99 >>> decision.value

100 10

101 >>> decision.freeze() # decision cant be changed afterwards

102 >>> decision.value = 12 # value cant be changed, will raise AssertionError

103 Traceback (most recent call last):

104 AssertionError: Can't change value of frozen decision

105 >>> decision.freeze(False) # unfreeze decision

106 >>> decision.reset() # reset to initial state

107 >>> decision.skip() # set value to None, only works if allow_skip flag set

108 """

109

110 SKIP = "skip"

111 SKIPALL = "skip all"

112 CANCEL = "cancel"

113 options = [SKIP, SKIPALL, CANCEL]

114

115 def __init__(self, question: str, console_identifier: str = None,

116 validate_func: Callable = None,

117 key: str = None, global_key: str = None,

118 allow_skip=False, validate_checksum=None,

119 related: List[str] = None, context: List[str] = None,

120 default=None, group: str = None,

121 representative_global_keys: list = None):

122

123 self.status = Status.pending

124 self._frozen = False

125 self._value = None

126

127 self.question = question

128 self.console_identifier = console_identifier

129 self.validate_func = validate_func

130 self.default = None

131 if default is not None:

132 if self.validate(default):

133 self.default = default

134 else:

135 logger.warning("Invalid default value (%s) for %s: %s",

136 default, self.__class__.__name__, self.question)

137

138 self.key = key

139 self.global_key = global_key

140

141 self.allow_skip = allow_skip

142 self.allow_save_load = bool(global_key)

143

144 self.validate_checksum = validate_checksum

145

146 self.related = related

147 self.context = context

148

149 self.group = group

150 self.representative_global_keys = representative_global_keys

151

152 @property

153 def value(self):

154 """Answer value of decision.

155

156 Raises:

157 ValueError: On accessing the value before it is set or by setting an invalid value

158 AssertionError: By changing a frozen Decision

159 """

160 if self.valid():

161 return self._value

162 else:

163 raise ValueError("Can't get value from invalid decision.")

164

165 @value.setter

166 def value(self, value):

167 if self._frozen:

168 raise AssertionError("Can't change value of frozen decision")

169 # if self.status != Status.pending:

170 # raise ValueError("Decision is not pending. Call reset() first.")

171 _value = self.convert(value)

172 if _value is None:

173 self.skip()

174 elif self.validate(_value):

175 self._value = _value

176 self.status = Status.ok

177 else:

178 raise ValueError("Invalid value: %r for %s" % (value, self.question))

179

180 def reset(self):

181 """Reset the Decision to it's initial state.

182

183 Raises:

184 AssertionError: if Decision is frozen

185 """

186 if self._frozen:

187 raise AssertionError("Can't change frozen decision")

188 self.status = Status.pending

189 self._value = None

190

191 def freeze(self, freeze=True):

192 """Freeze this Decision to prevent further manipulation.

193

194 Args:

195 freeze: the freeze state

196

197 Raises:

198 AssertionError: If the Decision is currently pending

199 """

200 if self.status == Status.pending and freeze:

201 raise AssertionError(

202 "Can't freeze pending decision. Set valid value first.")

203 self._frozen = freeze

204

205 def skip(self):

206 """Set value to None und mark as solved."""

207 if not self.allow_skip:

208 raise DecisionException("This Decision can not be skipped.")

209 if self._frozen:

210 raise DecisionException("Can't change frozen decision.")

211 if self.status != Status.pending:

212 raise DecisionException(

213 "This Decision is not pending. Call reset() first.")

214 self._value = None

215 self.status = Status.skipped

216

217 @staticmethod

218 def build_checksum(item):

219 """Create checksum for item."""

220 return hashlib.md5(json.dumps(item, sort_keys=True)

221 .encode('utf-8')).hexdigest()

222

223 def convert(self, value):

224 """Convert value to inner type."""

225 return value

226

227 def _validate(self, value):

228 raise NotImplementedError("Implement method _validate!")

229

230 def validate(self, value) -> bool:

231 """Checks value with validate_func and returns truth value."""

232 _value = self.convert(value)

233 basic_valid = self._validate(_value)

234

235 if self.validate_func:

236 if type(self.validate_func) is not list:

237 self.validate_func = [self.validate_func]

238 check_list = []

239 for fnc in self.validate_func:

240 try:

241 check_list.append(bool(fnc(_value)))

242 except:

243 check_list.append(False)

244 external_valid = all(check_list)

245 else:

246 external_valid = True

247

248 return basic_valid and external_valid

249

250 def valid(self) -> bool:

251 """Check if Decision is valid."""

252 return self.status == Status.ok \

253 or (self.status == Status.skipped and self.allow_skip)

254

255 def reset_from_deserialized(self, kwargs):

256 """Reset decision from its serialized form."""

257 value = kwargs['value']

258 checksum = kwargs.get('checksum')

259 if value is None:

260 return

261 valid = False

262 if self.validate_func:

263 if type(self.validate_func) is not list:

264 self.validate_func = [self.validate_func]

265 check_list = []

266 for fnc in self.validate_func:

267 check_list.append(bool(fnc(value)))

268 valid = all(check_list)

269 if (not self.validate_func) or valid:

270 if checksum == self.validate_checksum:

271 self.value = self.deserialize_value(value)

272 self.status = Status.ok

273 logger.info("Loaded decision '%s' with value: %s", self.global_key, value)

274 else:

275 logger.warning("Checksum mismatch for loaded decision '%s", self.global_key)

276 else:

277 logger.warning("Check for loaded decision '%s' failed. Loaded value: %s",

278 self.global_key, value)

279

280 def serialize_value(self):

281 """Return JSON serializable value."""

282 return {'value': self.value}

283

284 def deserialize_value(self, value):

285 """rebuild value from json deserialized object"""

286 return value

287

288 def get_serializable(self):

289 """Returns json serializable object representing state of decision"""

290 kwargs = self.serialize_value()

291 if self.validate_checksum:

292 kwargs['checksum'] = self.validate_checksum

293 return kwargs

294

295 def get_options(self):

296 """Get all available options."""

297 options = [Decision.CANCEL]

298 if self.allow_skip:

299 options.append(Decision.SKIP)

300

301 return options

302

303 def get_question(self) -> str:

304 """Get the question."""

305 return self.question

306

307 def get_body(self):

308 """Returns list of tuples representing items of CollectionDecision else None"""

309 return None

310

311 def __repr__(self):

312 value = str(self.value) if self.status == Status.ok else '???'

313 return '<%s (<%s> Q: "%s" A: %s)>' % (

314 self.__class__.__name__, self.status, self.question, value)

315

316

317class RealDecision(Decision):

318 """Accepts input of type real.

319

320 Args:

321 unit: the unit of the Decisions value

322 """

323

324 def __init__(self, *args, unit: pint.Quantity = None, **kwargs):

325 self.unit = unit if unit else ureg.dimensionless

326 default = kwargs.get('default')

327 if default is not None and not isinstance(default, pint.Quantity):

328 kwargs['default'] = default * self.unit

329 super().__init__(*args, **kwargs)

330

331 def convert(self, value):

332 if not isinstance(value, pint.Quantity):

333 try:

334 return value * self.unit

335 except:

336 pass

337 return value

338

339 def _validate(self, value):

340 if isinstance(value, pint.Quantity):

341 try:

342 float(value.m)

343 except:

344 pass

345 else:

346 return True

347 return False

348

349 def get_question(self):

350 return "{} in [{}]".format(self.question, self.unit)

351

352 def get_body(self):

353 return {'unit': str(self.unit)}

354

355 def get_debug_answer(self):

356 answer = super().get_debug_answer()

357 if isinstance(answer, pint.Quantity):

358 return answer.to(self.unit)

359 return answer * self.unit

360

361 def serialize_value(self):

362 kwargs = {

363 'value': self.value.magnitude,

364 'unit': str(self.value.units)

365 }

366 return kwargs

367

368 def reset_from_deserialized(self, kwargs):

369 kwargs['value'] = kwargs['value'] * ureg[kwargs.pop('unit', str(self.unit))]

370 super().reset_from_deserialized(kwargs)

371

372

373class BoolDecision(Decision):

374 """Accepts input convertable as bool"""

375

376 POSITIVES = ("y", "yes", "ja", "j", "1")

377 NEGATIVES = ("n", "no", "nein", "n", "0")

378

379 def __init__(self, *args, **kwargs):

380 super().__init__(*args, validate_func=None, **kwargs)

381

382 @staticmethod

383 def _validate(value):

384 """validates if value is acceptable as bool"""

385 return value is True or value is False

386

387

388class ListDecision(Decision):

389 """Accepts index of list element as input.

390

391 Args:

392 choices: a list of values where str(value) is used for labels or a list of (value, label) tuples

393 live_search:

394

395 """

396

397 def __init__(self, *args, choices:List[Union[Any, Tuple[Any, str]]], live_search=False, **kwargs):

398 if not choices:

399 raise AttributeError("choices must hold at least one item")

400 if hasattr(choices[0], '__len__') and len(choices[0]) == 2:

401 self.items = [choice[0] for choice in choices]

402 self.labels = [str(choice[1]) for choice in choices]

403 else:

404 self.items = choices

405 # self.labels = [str(choice) for choice in self.items]

406

407 self.live_search = live_search

408 super().__init__(*args, validate_func=None, **kwargs)

409

410 if len(self.items) == 1:

411 if not self.status != Status.pending:

412 # set only item as default

413 if self.default is None:

414 self.default = self.items[0]

415

416 @property

417 def choices(self):

418 """Available choices for the Decision."""

419 if hasattr(self, 'labels'):

420 return zip(self.items, self.labels)

421 else:

422 return self.items

423

424 def _validate(self, value):

425 pass # _validate not required. see validate

426

427 def validate(self, value):

428 return value in self.items

429

430 def get_body(self):

431 body = []

432 for i, item in enumerate(self.choices):

433 if isinstance(item, (list, tuple)) and len(item) == 2:

434 # label provided

435 body.append((i, *item))

436 else:

437 # no label provided

438 body.append((i, item, ' '))

439 return body

440

441

442class StringDecision(Decision):

443 """Accepts string input"""

444

445 def __init__(self, *args, min_length=1, **kwargs):

446 self.min_length = min_length

447 super().__init__(*args, **kwargs)

448

449 def _validate(self, value):

450 return isinstance(value, str) and len(value) >= self.min_length

451

452

453class GuidDecision(Decision):

454 """Accepts GUID(s) as input. Value is a set of GUID(s)"""

455

456 def __init__(self, *args, multi=False, **kwargs):

457 self.multi = multi

458 super().__init__(*args, **kwargs)

459

460 def _validate(self, value):

461 if isinstance(value, set) and value:

462 if not self.multi and len(value) != 1:

463 return False

464 return all(isinstance(guid, str) and len(guid) == 22 for guid in value)

465 return False

466

467 def serialize_value(self):

468 return {'value': list(self.value)}

469

470 def deserialize_value(self, value):

471 return set(value)

472

473

474class DecisionBunch(list):

475 """Collection of decisions."""

476

477 def __init__(self, decisions: Iterable[Decision] = ()):

478 super().__init__(decisions)

479

480 def valid(self) -> bool:

481 """Check status of all decisions."""

482 return all(decision.status in (Status.ok, Status.skipped)

483 for decision in self)

484

485 def to_answer_dict(self) -> Dict[Any, Decision]:

486 """Create dict from DecisionBunch using decision.key."""

487 return {decision.key: decision.value for decision in self}

488

489 def to_serializable(self) -> dict:

490 """Create JSON serializable dict of decisions."""

491 decisions = {decision.global_key: decision.get_serializable()

492 for decision in self}

493 return decisions

494

495 def validate_global_keys(self):

496 """Check if all global keys are unique.

497

498 :raises: AssertionError on bad keys."""

499 # mapping = {decision.global_key: decision for decision in self}

500 count = Counter(item.global_key for item in self if item.global_key)

501 duplicates = {decision for (decision, v) in count.items() if v > 1}

502

503 if duplicates:

504 raise AssertionError("Following global keys are not unique: %s",

505 duplicates)

506

507 def get_reduced_bunch(self, criteria: str = 'key'):

508 """Reduces the decisions to one decision per unique key.

509

510 To reduce the number of decisions in some cases the same answer can be

511 used for multiple decisions. This method allows to reduce the number

512 of decisions based on a given criteria.

513

514 Args:

515 criteria: criteria based on which the decisions should be reduced.

516 Possible are 'key' and 'question'.

517 Returns:

518 unique_decisions: A DecisionBunch with only unique decisions based

519 on criteria

520 """

521 pos_criteria = ['key', 'question']

522 if criteria not in pos_criteria:

523 raise NotImplementedError(f'Pick one of these valid options:'

524 f' {pos_criteria}')

525 unique_decisions = DecisionBunch()

526 doubled_decisions = DecisionBunch()

527 existing_criteria = []

528 for decision in self:

529 cur_key = getattr(decision, criteria)

530 if cur_key not in existing_criteria:

531 unique_decisions.append(decision)

532 existing_criteria.append(cur_key)

533 else:

534 doubled_decisions.append(decision)

535

536 return unique_decisions, doubled_decisions

537

538

539def save(bunch: DecisionBunch, path):

540 """Save solved Decisions to file system"""

541

542 decisions = bunch.to_serializable()

543 data = {

544 'version': version("bim2sim"),

545 'checksum_ifc': None,

546 'decisions': decisions,

547 }

548 with open(path, "w") as file:

549 json.dump(data, file, indent=2)

550 logger.info("Saved %d decisions.", len(bunch))

551

552

553def load(path) -> Dict[str, Any]:

554 """Load previously solved Decisions from file system."""

555

556 try:

557 with open(path, "r") as file:

558 data = json.load(file)

559 except IOError as ex:

560 logger.info(f"Unable to load decisions. "

561 f"No Existing decisions found at {ex.filename}")

562 return {}

563 cur_version = data.get('version', '0')

564 if cur_version != version("bim2sim"):

565 try:

566 data = convert(cur_version, version("bim2sim"), data)

567 logger.info("Converted stored decisions from version '%s' to '%s'",

568 cur_version, version("bim2sim"))

569 except:

570 logger.error("Decision conversion from %s to %s failed")

571 return {}

572 decisions = data.get('decisions')

573 logger.info("Found %d previous made decisions.", len(decisions or []))

574 return decisions

Coverage for bim2sim/kernel/decision/init.py: 88%

299 statements