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