251 lines
No EOL
9.7 KiB
Python
251 lines
No EOL
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .profiles import Profile
|
|
|
|
class LocalCache:
|
|
profiles: dict[str, Profile] = {}
|
|
|
|
files_metadata: dict[str, Any] = {}
|
|
|
|
@property
|
|
def display_profiles(self) -> dict[str, Profile]:
|
|
return {k: profile for k, profile in self.profiles.items() if '*' not in profile.id}
|
|
|
|
@property
|
|
def printers_profiles(self) -> dict[str, Profile]:
|
|
return {k: profile for k, profile in self.display_profiles.items() if profile.category == 'printer'}
|
|
|
|
@property
|
|
def print_profiles(self) -> dict[str, Profile]:
|
|
return {k: profile for k, profile in self.display_profiles.items() if profile.category == 'print'}
|
|
|
|
@property
|
|
def filament_profiles(self) -> dict[str, Profile]:
|
|
return {k: profile for k, profile in self.display_profiles.items() if profile.category == 'filament'}
|
|
|
|
def load(self, dirs: list[Path | str]) -> tuple[dict[str, tuple[Any, Any]], dict[str, Any], dict[str, Any]]:
|
|
|
|
old = self.files_metadata.copy()
|
|
self._fetch_files_metadata(dirs)
|
|
new = self.files_metadata
|
|
|
|
changed = {k: (old[k], new[k]) for k in old.keys() & new.keys() if old[k] != new[k]}
|
|
added = {k: new[k] for k in new.keys() - old.keys()}
|
|
deleted = {k: old[k] for k in old.keys() - new.keys()}
|
|
|
|
if len(changed | added | deleted) == 0: return {}, {}, {}
|
|
|
|
for deleted_file in deleted:
|
|
keys_to_remove = [key for key, val in self.profiles.items() if str(val.path) == deleted_file]
|
|
for key in keys_to_remove:
|
|
self.profiles.pop(key, None)
|
|
|
|
for file_path in (changed | added):
|
|
self._process_ini_to_cache_dict(file_path)
|
|
|
|
self.files_metadata = new
|
|
|
|
for k, profile in self.display_profiles.items():
|
|
profile.generate_inherited_confs(self.profiles)
|
|
|
|
return changed, added, deleted
|
|
|
|
@property
|
|
def vendors(self):
|
|
return sorted({ p.vendor for p in self.filament_profiles.values() })
|
|
|
|
def evaluate_compatibility(self, enabled_printers: set[str], enabled_vendors: set[str]):
|
|
for k, profile in self.printers_profiles.items():
|
|
if k not in enabled_printers: continue
|
|
# if profile.compatible_profiles: continue
|
|
enabled_vendors.add("")
|
|
enabled_filament_profiles = {k: p for k, p in self.filament_profiles.items() if p.vendor in enabled_vendors | {''}}
|
|
profile.evaluate_compatibility({k: pp.compatibility_expression for k, pp in (enabled_filament_profiles | self.print_profiles).items()})
|
|
|
|
def _fetch_files_metadata(self, dirs: list[Path | str]):
|
|
self.files_metadata = {}
|
|
# Iterate over all provided directories
|
|
for directory in dirs:
|
|
sanitized_path = self._sanitize_directory(directory)
|
|
if not sanitized_path:
|
|
continue
|
|
|
|
# Use os.walk with followlinks=True to ensure linked folders are processed
|
|
for root, _, files in os.walk(sanitized_path, followlinks=True):
|
|
for file in files:
|
|
if file.endswith('.ini'):
|
|
file_path = Path(root) / file
|
|
try:
|
|
last_modified = file_path.stat().st_mtime
|
|
self.files_metadata[str(file_path)] = last_modified
|
|
except OSError as e:
|
|
print(f"Error reading file {file_path}: {e}")
|
|
continue
|
|
|
|
def _sanitize_directory(self, dir_str: Path | str) -> Path | None:
|
|
if not dir_str:
|
|
return None
|
|
|
|
sanitized = Path(os.path.expanduser(dir_str)).resolve()
|
|
|
|
if not sanitized.is_dir():
|
|
print(f"Path is not a valid directory: {dir_str}")
|
|
return None
|
|
|
|
return sanitized
|
|
|
|
def _process_ini_to_cache_dict(self, path: str):
|
|
# Convert ConfigParser content into a dictionary
|
|
has_header, ini_dict = ini_to_dict(path)
|
|
|
|
# Flatten the dictionary for profiles and add to self.config_headers
|
|
for key, conf_dict in ini_dict.items():
|
|
if ":" in key:
|
|
self.profiles[key] = Profile(
|
|
key,
|
|
key.split(':')[0] if len(key.split(':')) > 1 else '',
|
|
Path(path),
|
|
has_header,
|
|
conf_dict
|
|
)
|
|
|
|
return
|
|
|
|
def generate_conf_writer(self, printer_profile: str, filament_profile: list[str], print_profile: str, overrides: dict[str, dict[str, str]]) -> 'ConfigWriter':
|
|
from .prusaslicer_fields import search_db
|
|
|
|
conf = {}
|
|
|
|
# add printer and print profile
|
|
conf.update(self.profiles[printer_profile].all_conf_dict)
|
|
conf.update(self.profiles[print_profile].all_conf_dict)
|
|
|
|
# add filament profiles per extruder
|
|
filament_merged_conf: dict[str, str] = {}
|
|
filament_confs: list[dict[str, str]] = [self.profiles[profile].all_conf_dict for profile in filament_profile]
|
|
common_keys: set[str] = set().union(*filament_confs)
|
|
common_keys.discard('cooling_slowdown_logic')
|
|
|
|
for key in common_keys:
|
|
key_props: dict[str, Any] | None = search_db.get(key, None)
|
|
|
|
if not key_props: print(f'Key {key} unrecognized! Check your configuration'); continue
|
|
|
|
default_value: str = key_props.get('default', '')
|
|
key_type: str = key_props['type']
|
|
separator = ',' if key_type in ['coPercents', 'coFloats', 'coFloatsOrPercents', 'coInts', 'coIntsNullable', 'coBools', 'coPoints'] else ';'
|
|
|
|
values: list[str] = [
|
|
d.get(key, default_value)
|
|
for d in filament_confs
|
|
]
|
|
|
|
joined: str = separator.join(values)
|
|
filament_merged_conf[key] = joined
|
|
|
|
conf.update(filament_merged_conf)
|
|
|
|
# overwrite with overrides
|
|
conf.update({k: o['value'] for k, o in overrides.items()})
|
|
|
|
# add profile names
|
|
conf.update({
|
|
'printer_settings_id': printer_profile.split(":")[1],
|
|
'filament_settings_id': ";".join(p.split(":")[1] for p in filament_profile),
|
|
'print_settings_id': print_profile.split(":")[1],
|
|
})
|
|
|
|
return ConfigWriter(conf)
|
|
|
|
class ConfigWriter:
|
|
def __init__(self, conf: dict[str, str | list[str]]) -> None:
|
|
self.config_dict: dict[str, str | list[str]] = conf
|
|
self.temp_dir = tempfile.gettempdir()
|
|
|
|
def write_ini_3mf(self, config_local_path: Path):
|
|
with open(config_local_path, 'w') as file:
|
|
for key, val in dict(sorted(self.config_dict.items())).items():
|
|
file.write(f"; {key} = {val}\n")
|
|
|
|
def get(self, key: str, default: Any = None) -> str | list[str]:
|
|
return self.config_dict[key]
|
|
|
|
@property
|
|
def checksum(self) -> int:
|
|
import json, zlib
|
|
data = json.dumps(self.config_dict, sort_keys=True).encode("utf-8")
|
|
return zlib.crc32(data) & 0xFFFFFFFF
|
|
|
|
def generate_conf(profiles: dict[str, Any], id: str) -> dict[str, str]:
|
|
if not (profile := profiles.get(id)): return {}
|
|
if not profile.conf_dict: return {}
|
|
conf_current = profiles[id].conf_dict # Copy to avoid modifying the original config
|
|
if conf_current.get('inherits', False):
|
|
curr_category = id.split(":")[0]
|
|
inherited_ids = [curr_category + ":" + inherit_id.strip() for inherit_id in conf_current['inherits'].split(';')] # Split on semicolon for multiple inheritance
|
|
merged_conf = {}
|
|
for inherit_id in inherited_ids:
|
|
if inherit_id in profiles:
|
|
inherited_conf = generate_conf(profiles, inherit_id) # Recursive call for each inherited config
|
|
merged_conf.update(inherited_conf) # Merge each inherited config
|
|
merged_conf.update(conf_current) # Update with current config values (overriding inherited)
|
|
conf_current = merged_conf
|
|
conf_current.pop('inherits', None)
|
|
conf_current.pop('renamed_from', None)
|
|
return conf_current
|
|
|
|
import re
|
|
from configparser import ConfigParser, MissingSectionHeaderError
|
|
|
|
def ini_to_dict(path: str) -> tuple[bool, dict[str, dict[str, str]]]:
|
|
# Read the file content from the path
|
|
with open(path, 'r') as file:
|
|
content = file.read()
|
|
|
|
config = ConfigParser(interpolation=None)
|
|
try:
|
|
# Attempt to parse the content
|
|
config.read_string(content)
|
|
has_header = True
|
|
|
|
except MissingSectionHeaderError:
|
|
# Determine category based on specific IDs in the content
|
|
if re.search(r'^filament_settings_id', content, re.MULTILINE):
|
|
cat = 'filament'
|
|
elif re.search(r'^print_settings_id', content, re.MULTILINE):
|
|
cat = 'print'
|
|
elif re.search(r'^printer_settings_id', content, re.MULTILINE):
|
|
cat = 'printer'
|
|
else:
|
|
raise ValueError(f"Unable to determine category for the INI file: {path}")
|
|
|
|
# Extract the filename without extension
|
|
name = os.path.splitext(os.path.basename(path))[0]
|
|
# Create a default section with the determined category and name
|
|
default_section = f"[{cat}:{name}]\n" + content
|
|
config.read_string(default_section)
|
|
has_header = False
|
|
|
|
# Convert ConfigParser content into a dictionary
|
|
return has_header, {
|
|
section: dict(sorted(config.items(section)))
|
|
for section in sorted(config.sections())
|
|
}
|
|
|
|
def ini_content_to_dict(path: str) -> dict[str, str]:
|
|
# Read the file content from the path
|
|
with open(path, 'r') as file:
|
|
content = file.read()
|
|
|
|
config = ConfigParser(interpolation=None)
|
|
|
|
default_section = f"[default:default]\n" + content
|
|
config.read_string(default_section)
|
|
|
|
return dict(sorted(config.items(config.sections()[0]))) |