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