Coverage for bim2sim/project.py: 74%
301 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-17 11:04 +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).value
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)
328 self.plugin_cls = self._get_plugin(plugin)
329 self.playground = Playground(self)
330 self.sim_settings = self.playground.sim_settings
332 def _get_plugin(self, plugin):
333 if plugin and isinstance(plugin, str):
334 return load_plugin(plugin)
335 elif plugin and issubclass(plugin, Plugin):
336 return plugin
337 else:
338 plugin_name = self.config['Backend']['use']
339 assert plugin_name, "Either an explicit passed plugin or" \
340 " equivalent entry in config is required."
341 return load_plugin(plugin_name)
343 @classmethod
344 def create(cls, project_folder, ifc_paths: Dict = None, plugin: Union[
345 str, Type[Plugin]] = None, open_conf: bool = False):
346 """Create new project
348 Args:
349 project_folder: directory of project
350 ifc_paths: dict with key: IFCDomain and value: path
351 to corresponding ifc which gets copied into project folder
352 plugin: Plugin to use with this project. If passed as string,
353 make sure it is importable (see plugins.load_plugin)
354 open_conf: flag to open the config file in default application
355 updated from config
356 """
357 # create folder first and use given plugin
358 if plugin and (isinstance(plugin, str) or issubclass(plugin, Plugin)):
359 FolderStructure.create(
360 project_folder, ifc_paths, plugin, open_conf)
361 project = cls(project_folder, plugin=plugin)
362 else:
363 # recreate plugin out of config, since no plugin was given
364 FolderStructure.create(
365 project_folder, ifc_paths, open_conf=open_conf)
366 project = cls(project_folder)
368 return project
370 @staticmethod
371 def is_project_folder(path: str) -> bool:
372 return FolderStructure(path).is_project_folder()
374 @classmethod
375 def _release(cls, project):
376 if cls._active_project is project:
377 cls._active_project = None
378 elif cls._active_project is None:
379 raise AssertionError("Cant release Project. No active project.")
380 else:
381 raise AssertionError("Cant release from other project.")
383 def is_active(self) -> bool:
384 """Return True if current project is active, False otherwise."""
385 return Project._active_project is self
387 def _update_logging_thread_filters(self):
388 """Update thread filters to current thread."""
389 thread_name = threading.current_thread().name
390 for thread_filter in self._log_thread_filters:
391 thread_filter.thread_name = thread_name
393 @property
394 def config(self):
395 """returns configparser instance. Basic config is done if file is not
396 present"""
397 config = configparser.ConfigParser(allow_no_value=True)
398 if not config.read(self.paths.config):
399 config_base_setup(self.paths.root)
400 config.read(self.paths.config)
401 return config
403 def rewrite_config(self):
404 # TODO this might need changes due to handling of enums
405 config = self.config
406 settings_manager = self.sim_settings.manager
407 for setting in settings_manager:
408 s = settings_manager.get(setting)
409 if isinstance(s.value, LOD):
410 val = s.value.value
411 else:
412 val = s.value
413 config[type(self.sim_settings).__name__][s.name] = str(val)
415 with open(self.paths.config, "w") as file:
416 config.write(file)
418 def run(self, interactive=False, cleanup=True):
419 """Run project.
421 Args:
422 interactive: if True the Task execution order is determined by
423 Decisions else its derived by plugin
424 cleanup: execute cleanup logic. Not doing this is only relevant for
425 debugging
427 Raises:
428 AssertionError: if project setup is broken or on invalid Decisions
429 """
430 if not self.paths.is_project_folder():
431 raise AssertionError("Project ist not set correctly!")
433 self.sim_settings.check_mandatory()
434 success = False
435 if interactive:
436 run = self._run_interactive
437 else:
438 run = self._run_default
439 try:
440 # First update log filters in case Project was created from
441 # different thread.
442 # Then update log filters for each iteration, which might get
443 # called by a different thread.
444 # Deeper down multithreading is currently not supported for logging
445 # and will result in a mess of log messages.
446 self._update_logging_thread_filters()
447 for decision_bunch in run():
448 yield decision_bunch
449 self._update_logging_thread_filters()
450 if not decision_bunch.valid():
451 raise AssertionError("Cant continue with invalid decisions")
452 for decision in decision_bunch:
453 decision.freeze()
454 self._made_decisions.extend(decision_bunch)
455 self._made_decisions.validate_global_keys()
456 success = True
457 except Exception as ex:
458 self.logger.exception(f"Something went wrong!: {ex}")
459 finally:
460 if cleanup:
461 self.finalize(success=success)
462 return 0 if success else -1
464 def _run_default(self, plugin=None):
465 """Execution of plugins default tasks"""
466 # run plugin default
467 plugin_cls = plugin or self.plugin_cls
468 _plugin = plugin_cls()
469 for task_cls in _plugin.default_tasks:
470 yield from self.playground.run_task(task_cls(self.playground))
472 def _run_interactive(self):
473 """Interactive execution of available ITasks"""
474 while True:
475 tasks_classes = {task.__name__: task for task in
476 self.playground.available_tasks()}
477 choices = [(name, task.__doc__) for name, task in
478 tasks_classes.items()]
479 task_infos = [f"{task.__module__}.{task.__name__}" for task in
480 tasks_classes.values()]
481 task_infos_sorted = sorted(task_infos)
482 task_key_base = ",".join(task_infos_sorted)
483 task_key_hash = hashlib.sha256(task_key_base.encode()).hexdigest()
484 global_key = f"_task_{task_key_hash}_decision"
485 task_decision = ListDecision(
486 "What shall we do?",
487 choices=choices,
488 global_key=global_key
489 )
490 yield DecisionBunch([task_decision])
491 task_name = task_decision.value
492 task_class = tasks_classes[task_name]
493 yield from self.playground.run_task(task_class(self.playground))
494 if task_class.final:
495 break
497 def finalize(self, success=False):
498 """cleanup method"""
500 # clean up run relics
501 # backup decisions
502 if not success:
503 pth = self.paths.root / 'decisions_backup.json'
504 save(self._made_decisions, pth)
505 self.logger.warning("Decisions are saved in '%s'. Rename file to "
506 "'decisions.json' to reuse them.", pth)
507 self.logger.error(f'Project "{self.name}" '
508 f'finished, but not successful')
510 else:
511 save(self._made_decisions, self.paths.decisions)
512 self.logger.info(f'Project Exports can be found under '
513 f'{self.paths.export}')
514 self.logger.info(f'Project "{self.name}" finished successful')
516 # reset sim_settings:
517 self.playground.sim_settings.load_default_settings()
518 # clean logger
519 log.teardown_loggers()
521 def delete(self):
522 """Delete the project."""
523 self.finalize(True)
524 self.paths.delete(False)
525 self.logger.info("Project deleted")
527 def reset(self):
528 """Reset the current project."""
529 self.logger.info("Project reset")
530 self.playground.state.clear()
531 self.playground.history.clear()
532 self._made_decisions.clear()
534 def __repr__(self):
535 return "<Project(%s)>" % self.paths.root