Coverage for bim2sim/project.py: 74%
301 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-01 10:24 +0000
1"""Project handling"""
2import hashlib
3import logging
4import os
5import sys
6import subprocess
7import shutil
8import threading
9from distutils.dir_util import copy_tree
10from enum import Enum
11from pathlib import Path
12from typing import Dict, Type, Union
14import configparser
16from bim2sim.kernel.decision import ListDecision, DecisionBunch, save, load
17from bim2sim.kernel import log
18from bim2sim.tasks.base import Playground
19from bim2sim.plugins import Plugin, load_plugin
20from bim2sim.utilities.common_functions import all_subclasses
21from bim2sim.sim_settings import BaseSimSettings
22from bim2sim.utilities.types import LOD
25def open_config(path):
26 """Open config for user and wait for closing before continue."""
27 if sys.platform.startswith('darwin'): # For MAC OS X
28 open_file = subprocess.Popen(['open', path])
29 elif os.name == 'nt': # For Windows
30 open_file = subprocess.Popen(["notepad.exe", path])
31 # os.system("start " + conf_path)
32 # todo for any reason wait() seems not to work on linux
33 # elif os.name == 'posix': # For Linux, Mac, etc.
34 # open_file = subprocess.Popen(['xdg-open', path])
35 else:
36 raise NotImplementedError('Only mac os and windows are '
37 'supported currently.')
38 open_file.wait()
41def add_config_section(
42 config: configparser.ConfigParser,
43 sim_settings: BaseSimSettings,
44 name: str) -> configparser.ConfigParser:
45 """Add a section to config with all attributes and default values."""
46 if name not in config._sections:
47 config.add_section(name)
48 attributes = [attr for attr in list(sim_settings.__dict__.keys())
49 if not callable(getattr(sim_settings, attr)) and not
50 attr.startswith('__')]
51 for attr in attributes:
52 default_value = getattr(sim_settings, attr).default
53 if isinstance(default_value, Enum):
54 default_value = str(default_value)
55 if not attr in config[name]:
56 config[name][attr] = str(default_value)
57 return config
60def config_base_setup(path, backend=None):
61 """Initial setup for config file"""
62 config = configparser.ConfigParser(allow_no_value=True)
63 config.read(path)
65 if not config.sections():
66 # add all default attributes from base workflow
67 config = add_config_section(
68 config, BaseSimSettings, "Generic Simulation Settings")
69 # add all default attributes from sub workflows
70 all_settings = all_subclasses(BaseSimSettings)
71 for flow in all_settings:
72 config = add_config_section(config, flow, flow.__name__)
74 # add general settings
75 config.add_section("Backend")
76 config["Backend"]["use"] = backend
77 config.add_section("Frontend")
78 config["Frontend"]["use"] = 'ConsoleFrontEnd'
79 config.add_section("Modelica")
80 config["Modelica"]["Version"] = "4.0"
82 with open(path, "w") as file:
83 config.write(file)
86class FolderStructure:
87 """Project related file and folder handling."""
89 CONFIG = "config.toml"
90 DECISIONS = "decisions.json"
91 FINDER = "finder"
92 IFC_BASE = "ifc"
93 LOG = "log"
94 EXPORT = "export"
96 _src_path = Path(__file__).parent # base path to bim2sim assets
98 def __init__(self, path=None):
99 self._root_path = None
100 self.root = path or os.getcwd()
102 @property
103 def root(self):
104 """absolute root path"""
105 return self._root_path
107 @root.setter
108 def root(self, value: str):
109 self._root_path = Path(value).absolute().resolve()
111 @property
112 def assets(self):
113 return self._src_path / 'assets'
115 @property
116 def enrichment(self):
117 return self._src_path / 'enrichment_data'
119 @property
120 def config(self):
121 """absolute path to config"""
122 return self._root_path / self.CONFIG
124 @property
125 def decisions(self):
126 """absolute path to decisions"""
127 return self._root_path / self.DECISIONS
129 @property
130 def finder(self):
131 """absolute path to finder"""
132 return self._root_path / self.FINDER
134 @property
135 def log(self):
136 """absolute path to log folder"""
137 return self._root_path / self.LOG
139 @property
140 def ifc_base(self):
141 """absolute path to ifc folder"""
142 return self._root_path / self.IFC_BASE
144 @property
145 def export(self):
146 """absolute path to export folder"""
147 return self._root_path / self.EXPORT
149 @property
150 def b2sroot(self):
151 """absolute path of bim2sim root folder"""
152 return self._src_path.parent
154 @property
155 def sub_dirs(self):
156 """list of paths to sub folders"""
157 return [self.log, self.ifc_base, self.export, self.finder]
159 def copy_assets(self, path):
160 """copy assets to project folder"""
161 copy_tree(str(self.assets), str(path))
163 def is_project_folder(self, path=None):
164 """Check if root path (or given path) is a project folder"""
165 root = path or self.root
166 if not root:
167 return False
168 root = Path(root)
169 if not root.is_dir():
170 return False
171 if (root / self.CONFIG).is_file():
172 return True
173 return False
175 def complete_project_folder(self):
176 """Adds missing sub folders to given path"""
177 for subdir in self.sub_dirs:
178 os.makedirs(subdir, exist_ok=True)
180 def create_project_folder(self):
181 """Creates a project folder on given path"""
182 os.makedirs(self.root, exist_ok=True)
183 self.complete_project_folder()
185 # create empty config file
186 with open(self.config, "w"):
187 pass
189 self.copy_assets(self.root)
191 @classmethod
192 def create(cls, rootpath: str, ifc_paths: Dict = None,
193 plugin: Union[str, Type[Plugin]] = None, open_conf: bool = False):
194 """Create ProjectFolder and set it up.
196 Create instance, set source path, create project folder
197 copy ifc, base config setup and open config if needed.
199 Args:
200 rootpath: path of root folder
201 ifc_paths: dict with key: bim2sim domain and value: path
202 to corresponding ifc which gets copied into project folder
203 plugin: the target Plugin
204 open_conf: flag to open the config file in default application
205 """
206 general_logger = log.initial_logging_setup()
207 # set rootpath
208 self = cls(rootpath)
209 if isinstance(plugin, str):
210 plugin_name = plugin
211 elif issubclass(plugin, Plugin):
212 plugin_name = plugin.name
213 else:
214 raise ValueError(f"{plugin} is not a subclass of Plugin or a str.")
216 if self.is_project_folder():
217 general_logger.info(
218 "Given path is already a project folder ('%s')" % self.root)
219 else:
220 self.create_project_folder()
221 config_base_setup(self.config, plugin_name)
223 if ifc_paths:
224 if not isinstance(ifc_paths, Dict):
225 raise ValueError(
226 "Please provide a Dictionary with key: Domain, value: Path "
227 "to IFC ")
228 # copy all ifc files to domain specific project folders
229 for domain, file_path in ifc_paths.items():
230 if not file_path.exists():
231 if "test" in file_path.parts \
232 and "resources" in file_path.parts:
233 raise ValueError(
234 f"Provided path to ifc is: {file_path}, but this "
235 f"file does not exist. You are trying to run a "
236 f"test on your local machine,"
237 f" but it seems like you have not downloaded the"
238 f" needed test resources. "
239 f"Run 'git submodule update --init --recursive' "
240 f"to make sure to include the test resources."
241 )
242 else:
243 raise ValueError(
244 f"Provided path to ifc is: {file_path}, but this file "
245 f"does not exist.")
246 Path.mkdir(self.ifc_base / domain.name, exist_ok=True)
247 shutil.copy2(
248 file_path, self.ifc_base / domain.name / file_path.name)
250 if open_conf:
251 # open config for user interaction
252 open_config(self.config)
253 general_logger.info("Project folder created.")
254 return self
256 def delete(self, confirm=True):
257 """Delete project folder and all files in it.
259 Raises:
260 AssertionError: if not existing on file system
261 """
262 # TODO: decision system
263 if confirm:
264 ans = input("Delete project folder and all included files? [y/n]")
265 if not ans == 'y':
266 return
268 if os.path.exists(self.root):
269 shutil.rmtree(self.root, ignore_errors=True)
270 print("Project folder deleted.")
271 else:
272 raise AssertionError(
273 "Can't delete project folder (reason: does not exist)")
275 def __str__(self):
276 return str(self.root)
278 def __repr__(self):
279 return "<FolderStructure (root: %s)>" % self.root
282class Project:
283 """Project resource handling.
285 Args:
286 path: path to load project from
287 plugin: Plugin to use. This overwrites plugin from config.
289 Raises:
290 AssertionError: on invalid path. E.g. if not existing
291 """
292 _active_project = None # lock to prevent multiple interfering projects
294 def __init__(
295 self,
296 path: str = None,
297 plugin: Type[Plugin] = None,
298 ):
299 """Load existing project"""
300 self.paths = FolderStructure(path)
301 self.thread_name = threading.current_thread().name
302 self.log_handlers, self._log_thread_filters = log.project_logging_setup(self)
303 self.logger = logging.getLogger('bim2sim')
304 # try to get name of project from ifc name
305 try:
306 # prioritize the name of the architectural model, otherwise the
307 # name of the project may be an arbitrary model name in
308 # multi-model projects
309 arch_name_list = [i.stem for i in list(
310 filter(Path.is_file, self.paths.ifc_base.glob('**/*')))
311 if 'arch' in i.as_posix()]
312 if arch_name_list:
313 self.name = arch_name_list[0]
314 else:
315 self.name = list(
316 filter(
317 Path.is_file, self.paths.ifc_base.glob('**/*')))[0].stem
318 except:
319 self.logger.warning(
320 'Could not set correct project name, using "Project"!')
321 self.name = "Project"
323 if not self.paths.is_project_folder():
324 raise AssertionError("Project path is no valid project directory. "
325 "Use Project.create() to create a new Project")
326 self._made_decisions = DecisionBunch()
327 self.loaded_decisions = load(self.paths.decisions)
329 self.plugin_cls = self._get_plugin(plugin)
330 self.playground = Playground(self)
331 # link sim_settings to project to make set of settings easier
332 self.sim_settings = self.playground.sim_settings
334 def _get_plugin(self, plugin):
335 if plugin and isinstance(plugin, str):
336 return load_plugin(plugin)
337 elif plugin and issubclass(plugin, Plugin):
338 return plugin
339 else:
340 plugin_name = self.config['Backend']['use']
341 assert plugin_name, "Either an explicit passed plugin or" \
342 " equivalent entry in config is required."
343 return load_plugin(plugin_name)
345 @classmethod
346 def create(cls, project_folder, ifc_paths: Dict = None, plugin: Union[
347 str, Type[Plugin]] = None, open_conf: bool = False):
348 """Create new project
350 Args:
351 project_folder: directory of project
352 ifc_paths: dict with key: IFCDomain and value: path
353 to corresponding ifc which gets copied into project folder
354 plugin: Plugin to use with this project. If passed as string,
355 make sure it is importable (see plugins.load_plugin)
356 open_conf: flag to open the config file in default application
357 updated from config
358 """
359 # create folder first and use given plugin
360 if plugin and (isinstance(plugin, str) or issubclass(plugin, Plugin)):
361 FolderStructure.create(
362 project_folder, ifc_paths, plugin, open_conf)
363 project = cls(project_folder, plugin=plugin)
364 else:
365 # recreate plugin out of config, since no plugin was given
366 FolderStructure.create(
367 project_folder, ifc_paths, open_conf=open_conf)
368 project = cls(project_folder)
370 return project
372 @staticmethod
373 def is_project_folder(path: str) -> bool:
374 return FolderStructure(path).is_project_folder()
376 @classmethod
377 def _release(cls, project):
378 if cls._active_project is project:
379 cls._active_project = None
380 elif cls._active_project is None:
381 raise AssertionError("Cant release Project. No active project.")
382 else:
383 raise AssertionError("Cant release from other project.")
385 def is_active(self) -> bool:
386 """Return True if current project is active, False otherwise."""
387 return Project._active_project is self
389 def _update_logging_thread_filters(self):
390 """Update thread filters to current thread."""
391 thread_name = threading.current_thread().name
392 for thread_filter in self._log_thread_filters:
393 thread_filter.thread_name = thread_name
395 @property
396 def config(self):
397 """returns configparser instance. Basic config is done if file is not
398 present"""
399 config = configparser.ConfigParser(allow_no_value=True)
400 if not config.read(self.paths.config):
401 config_base_setup(self.paths.root)
402 config.read(self.paths.config)
403 return config
405 def rewrite_config(self):
406 # TODO this might need changes due to handling of enums
407 config = self.config
408 settings_manager = self.sim_settings.manager
409 for setting in settings_manager:
410 s = settings_manager.get(setting)
411 if isinstance(s.value, LOD):
412 val = s.value.value
413 else:
414 val = s.value
415 config[type(self.sim_settings).__name__][s.name] = str(val)
417 with open(self.paths.config, "w") as file:
418 config.write(file)
420 def run(self, interactive=False, cleanup=True):
421 """Run project.
423 Args:
424 interactive: if True the Task execution order is determined by
425 Decisions else its derived by plugin
426 cleanup: execute cleanup logic. Not doing this is only relevant for
427 debugging
429 Raises:
430 AssertionError: if project setup is broken or on invalid Decisions
431 """
432 if not self.paths.is_project_folder():
433 raise AssertionError("Project ist not set correctly!")
435 self.sim_settings.check_mandatory()
436 success = False
437 if interactive:
438 run = self._run_interactive
439 else:
440 run = self._run_default
441 try:
442 # First update log filters in case Project was created from
443 # different thread.
444 # Then update log filters for each iteration, which might get
445 # called by a different thread.
446 # Deeper down multithreading is currently not supported for logging
447 # and will result in a mess of log messages.
448 self._update_logging_thread_filters()
449 for decision_bunch in run():
450 yield decision_bunch
451 self._update_logging_thread_filters()
452 if not decision_bunch.valid():
453 raise AssertionError("Cant continue with invalid decisions")
454 for decision in decision_bunch:
455 decision.freeze()
456 self._made_decisions.extend(decision_bunch)
457 self._made_decisions.validate_global_keys()
458 success = True
459 except Exception as ex:
460 self.logger.exception(f"Something went wrong!: {ex}")
461 finally:
462 if cleanup:
463 self.finalize(success=success)
464 return 0 if success else -1
466 def _run_default(self, plugin=None):
467 """Execution of plugins default tasks"""
468 # run plugin default
469 plugin_cls = plugin or self.plugin_cls
470 _plugin = plugin_cls()
471 for task_cls in _plugin.default_tasks:
472 yield from self.playground.run_task(task_cls(self.playground))
474 def _run_interactive(self):
475 """Interactive execution of available ITasks"""
476 while True:
477 tasks_classes = {task.__name__: task for task in
478 self.playground.available_tasks()}
479 choices = [(name, task.__doc__) for name, task in
480 tasks_classes.items()]
481 task_infos = [f"{task.__module__}.{task.__name__}" for task in
482 tasks_classes.values()]
483 task_infos_sorted = sorted(task_infos)
484 task_key_base = ",".join(task_infos_sorted)
485 task_key_hash = hashlib.sha256(task_key_base.encode()).hexdigest()
486 global_key = f"_task_{task_key_hash}_decision"
487 task_decision = ListDecision(
488 "What shall we do?",
489 choices=choices,
490 global_key=global_key
491 )
492 yield DecisionBunch([task_decision])
493 task_name = task_decision.value
494 task_class = tasks_classes[task_name]
495 yield from self.playground.run_task(task_class(self.playground))
496 if task_class.final:
497 break
499 def finalize(self, success=False):
500 """cleanup method"""
502 # clean up run relics
503 # backup decisions
504 if not success:
505 pth = self.paths.root / 'decisions_backup.json'
506 save(self._made_decisions, pth)
507 self.logger.warning("Decisions are saved in '%s'. Rename file to "
508 "'decisions.json' to reuse them.", pth)
509 self.logger.error(f'Project "{self.name}" '
510 f'finished, but not successful')
512 else:
513 save(self._made_decisions, self.paths.decisions)
514 self.logger.info(f'Project Exports can be found under '
515 f'{self.paths.export}')
516 self.logger.info(f'Project "{self.name}" finished successful')
518 # reset sim_settings:
519 self.playground.sim_settings.load_default_settings()
520 # clean logger
521 log.teardown_loggers()
523 def delete(self):
524 """Delete the project."""
525 self.finalize(True)
526 self.paths.delete(False)
527 self.logger.info("Project deleted")
529 def reset(self):
530 """Reset the current project."""
531 self.logger.info("Project reset")
532 self.playground.state.clear()
533 self.playground.history.clear()
534 self._made_decisions.clear()
536 def __repr__(self):
537 return "<Project(%s)>" % self.paths.root