165 lines
6.1 KiB
Python
165 lines
6.1 KiB
Python
from __future__ import annotations
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .models import Profile, ConfigWriter
|
|
from .utils import ini_to_dict
|
|
|
|
class LocalCache:
|
|
def __init__(self):
|
|
self.profiles: dict[str, Profile] = {}
|
|
self.files_metadata: dict[str, float] = {}
|
|
|
|
@property
|
|
def display_profiles(self) -> dict[str, Profile]:
|
|
"""Profiles that are not hidden (don't contain '*' in their ID)."""
|
|
return {k: p for k, p in self.profiles.items() if '*' not in p.id}
|
|
|
|
@property
|
|
def printer_profiles(self) -> dict[str, Profile]:
|
|
return {k: p for k, p in self.display_profiles.items() if p.category == 'printer'}
|
|
|
|
@property
|
|
def print_profiles(self) -> dict[str, Profile]:
|
|
return {k: p for k, p in self.display_profiles.items() if p.category == 'print'}
|
|
|
|
@property
|
|
def filament_profiles(self) -> dict[str, Profile]:
|
|
return {k: p for k, p in self.display_profiles.items() if p.category == 'filament'}
|
|
|
|
@property
|
|
def vendors(self) -> list[str]:
|
|
return sorted({p.vendor for p in self.filament_profiles.values() if p.vendor})
|
|
|
|
def load(self, dirs: list[Path | str]) -> tuple[set[str], set[str], set[str]]:
|
|
"""
|
|
Loads profiles from the specified directories.
|
|
Returns sets of (changed, added, deleted) file paths.
|
|
"""
|
|
old_metadata = self.files_metadata.copy()
|
|
new_metadata = self._fetch_files_metadata(dirs)
|
|
|
|
changed = {k for k in old_metadata.keys() & new_metadata.keys() if old_metadata[k] != new_metadata[k]}
|
|
added = new_metadata.keys() - old_metadata.keys()
|
|
deleted = old_metadata.keys() - new_metadata.keys()
|
|
|
|
if not (changed | added | deleted):
|
|
return set(), set(), set()
|
|
|
|
# Handle deletions
|
|
for deleted_file in deleted:
|
|
keys_to_remove = [k for k, v in self.profiles.items() if str(v.path) == deleted_file]
|
|
for k in keys_to_remove:
|
|
self.profiles.pop(k, None)
|
|
|
|
# Handle additions and changes
|
|
for file_path in (changed | added):
|
|
self._process_ini_to_cache(file_path)
|
|
|
|
self.files_metadata = new_metadata
|
|
|
|
# Recalculate inheritance for all profiles
|
|
for profile in self.display_profiles.values():
|
|
profile.generate_inherited_confs(self.profiles)
|
|
|
|
return changed, added, deleted
|
|
|
|
def evaluate_compatibility(self, enabled_printers: set[str], enabled_vendors: set[str]):
|
|
"""Evaluates compatibility expressions for all active profiles."""
|
|
for k, printer in self.printer_profiles.items():
|
|
if k not in enabled_printers:
|
|
continue
|
|
|
|
# Combine vendors with empty string for default/generic profiles
|
|
active_vendors = enabled_vendors | {''}
|
|
active_filaments = {k: p for k, p in self.filament_profiles.items() if p.vendor in active_vendors}
|
|
|
|
compats = {
|
|
k: p.compatibility_expression
|
|
for k, p in (active_filaments | self.print_profiles).items()
|
|
}
|
|
printer.evaluate_compatibility(compats)
|
|
|
|
def generate_conf_writer(
|
|
self,
|
|
printer_id: str,
|
|
filament_ids: list[str],
|
|
print_id: str,
|
|
overrides: dict[str, dict[str, Any]]
|
|
) -> ConfigWriter:
|
|
"""Merges selected profiles and overrides into a ConfigWriter."""
|
|
from .fields import search_db
|
|
|
|
conf: dict[str, Any] = {}
|
|
|
|
# Add printer and print profile
|
|
conf.update(self.profiles[printer_id].all_conf_dict)
|
|
conf.update(self.profiles[print_id].all_conf_dict)
|
|
|
|
# Merge filament profiles (handling multi-extruder setups)
|
|
filament_confs = [self.profiles[fid].all_conf_dict for fid in filament_ids]
|
|
all_keys = set().union(*filament_confs)
|
|
all_keys.discard('cooling_slowdown_logic')
|
|
|
|
for key in all_keys:
|
|
key_props = search_db.get(key)
|
|
if not key_props:
|
|
print(f"Warning: Key '{key}' unrecognized in fields database.")
|
|
continue
|
|
|
|
default_val = key_props.get('default', '')
|
|
key_type = key_props['type']
|
|
|
|
# Select separator based on field type
|
|
list_types = {
|
|
'coPercents', 'coFloats', 'coFloatsOrPercents',
|
|
'coInts', 'coIntsNullable', 'coBools', 'coPoints'
|
|
}
|
|
separator = ',' if key_type in list_types else ';'
|
|
|
|
values = [d.get(key, default_val) for d in filament_confs]
|
|
conf[key] = separator.join(map(str, values))
|
|
|
|
# Apply overrides
|
|
for k, o in overrides.items():
|
|
conf[k] = o['value']
|
|
|
|
# Add profile names for reference
|
|
conf.update({
|
|
'printer_settings_id': printer_id.split(":")[1],
|
|
'filament_settings_id': ";".join(fid.split(":")[1] for fid in filament_ids),
|
|
'print_settings_id': print_id.split(":")[1],
|
|
})
|
|
|
|
return ConfigWriter(conf)
|
|
|
|
def _fetch_files_metadata(self, dirs: list[Path | str]) -> dict[str, float]:
|
|
metadata = {}
|
|
for d in dirs:
|
|
path = Path(os.path.expanduser(d)).resolve()
|
|
if not path.is_dir():
|
|
continue
|
|
|
|
for root, _, files in os.walk(path, followlinks=True):
|
|
for f in files:
|
|
if f.endswith('.ini'):
|
|
file_path = Path(root) / f
|
|
try:
|
|
metadata[str(file_path)] = file_path.stat().st_mtime
|
|
except OSError:
|
|
continue
|
|
return metadata
|
|
|
|
def _process_ini_to_cache(self, path: str):
|
|
has_header, sections = ini_to_dict(path)
|
|
for key, conf in sections.items():
|
|
if ":" in key:
|
|
category = key.split(':')[0]
|
|
self.profiles[key] = Profile(
|
|
key=key,
|
|
category=category,
|
|
path=Path(path),
|
|
has_header=has_header,
|
|
conf_dict=conf
|
|
)
|