import os import stat import sys from pathlib import Path _INVOCATION = None BASE_DIR = Path(os.environ.get("SOLIXAUTO_HOME", Path.home() / "solix-automation")) DEVICE_PROFILE_DIR = BASE_DIR / "device-profiles" ANKER_PROFILE_DIR = DEVICE_PROFILE_DIR / "anker" SHELLY_PROFILE_DIR = DEVICE_PROFILE_DIR / "shelly" POWER_PROFILE_DIR = BASE_DIR / "power-profiles" STATE_DIR = BASE_DIR / "state" LOG_DIR = BASE_DIR / "logs" RUNTIME_STATE = STATE_DIR / "runtime.json" ENGINE_LOG = LOG_DIR / "automation.log" ALL_DIRS = [ BASE_DIR, DEVICE_PROFILE_DIR, ANKER_PROFILE_DIR, SHELLY_PROFILE_DIR, POWER_PROFILE_DIR, STATE_DIR, LOG_DIR, ] def ensure_dirs(): for directory in ALL_DIRS: directory.mkdir(parents=True, exist_ok=True) gitignore = BASE_DIR / ".gitignore" if not gitignore.exists(): gitignore.write_text( "device-profiles/\nstate/\nlogs/\n*.env\n.env\n", encoding="utf-8", ) return BASE_DIR def invocation(): global _INVOCATION if _INVOCATION is not None: return _INVOCATION script = sys.argv[0] if sys.argv else "" stem = Path(script).name if script else "" if stem in ("solixauto", "solixauto.exe"): _INVOCATION = "solixauto" return _INVOCATION if not script: _INVOCATION = "solixauto" return _INVOCATION interpreter = sys.executable or "python" display = script try: here = Path(script).resolve() if here.parent == Path.cwd(): display = here.name except (OSError, ValueError): pass _INVOCATION = f"{interpreter} {display}" return _INVOCATION def command(rest=""): base = invocation() return f"{base} {rest}".rstrip() def secure_file(path): path = Path(path) try: path.chmod(stat.S_IRUSR | stat.S_IWUSR) return True except (OSError, NotImplementedError): return False def permissions_enforced(): return not sys.platform.startswith("win") def relative(path): path = Path(path) try: return str(path.relative_to(BASE_DIR)) except ValueError: return str(path) def resolve_profile(reference, kind): reference = str(reference).strip() candidate = Path(reference).expanduser() if candidate.is_absolute() and candidate.exists(): return candidate roots = [BASE_DIR, DEVICE_PROFILE_DIR] if kind == "anker": roots.insert(0, ANKER_PROFILE_DIR) elif kind == "shelly": roots.insert(0, SHELLY_PROFILE_DIR) elif kind == "power": roots.insert(0, POWER_PROFILE_DIR) names = [reference] if not reference.endswith((".yaml", ".yml")): names.append(reference + ".yaml") names.append(reference + ".yml") for root in roots: for name in names: probe = root / name if probe.exists(): return probe return resolve_by_alias(reference, kind) def _slug(value): return "".join( char.lower() if char.isalnum() else "-" for char in str(value) ).strip("-") def resolve_by_alias(reference, kind): import yaml directories = [] if kind == "anker": directories.append(ANKER_PROFILE_DIR) elif kind == "shelly": directories.append(SHELLY_PROFILE_DIR) elif kind == "power": directories.append(POWER_PROFILE_DIR) else: directories.extend([ANKER_PROFILE_DIR, SHELLY_PROFILE_DIR, POWER_PROFILE_DIR]) wanted = _slug(reference) if not wanted: return None for directory in directories: if not directory.exists(): continue for candidate in sorted(directory.iterdir()): if candidate.suffix not in (".yaml", ".yml"): continue try: with candidate.open("r", encoding="utf-8") as handle: data = yaml.safe_load(handle) except Exception: continue if not isinstance(data, dict): continue haystack = list(data.get("aliases") or []) identity = data.get("identity") or {} for key in ("name", "id", "mac", "serial", "part_number"): if identity.get(key): haystack.append(identity[key]) access = data.get("access") or {} if access.get("host"): haystack.append(access["host"]) if data.get("name"): haystack.append(data["name"]) for entry in haystack: if _slug(entry) == wanted: return candidate return None