Merge branch 'main' of https://github.com/thisismygitrepo/crocodile

myresources/crocodile/cluster/session_managers.py

@@ -14,7 +14,8 @@
 class Zellij:
     @staticmethod
     def get_current_zellij_session() -> str:
-        try: return tb.L(tb.Terminal().run("zellij ls").op.split("\n")).filter(lambda x: "(current)" in x).list[0].replace(" (current)", "")
+        try:
+            return tb.L(tb.Terminal().run("zellij ls --no-formatting").op.split("\n")).filter(lambda x: "(current)" in x).list[0].split(" [Created")[0]
         except IndexError as ie:
             print(f"""Fails if there is no zellij session running, fails if there is no (current) suffix against the session name.""")
             raise ie
@@ -35,12 +36,10 @@ def open_console(ssh: Union[tb.SSH, SelfSSH], sess_name: str):
     @staticmethod
     def asssert_session_started(ssh: Union[tb.SSH, SelfSSH], sess_name: str):
         while True:
-            # if isinstance(ssh, SelfSSH):
-            # resp = tb.Terminal().run("zellij ls").op.split("\n")
-            # else:
-            resp = ssh.run("zellij ls", verbose=False).op.split("\n")
-            if sess_name in resp:
-                print(f"--> Session {resp} has started at the remote.")
+            raw_resp = ssh.run("zellij ls --no-formatting", verbose=False).op.split("\n")
+            current_sessions = [item.split(" [Created")[0] for item in raw_resp if "EXITED" not in item]
+            if sess_name in current_sessions:
+                print(f"--> Session {sess_name} has started at the remote.")
                 time.sleep(6)
                 break
             time.sleep(2)

myresources/crocodile/file_management.py

@@ -639,9 +639,10 @@ def __init__(self, source_func: Callable[[], 'T'], expire: Union[str, timedelta]
         self.expire = str2timedelta(expire) if isinstance(expire, str) else expire
         self.name = name if isinstance(name, str) else str(self.source_func)
     @property
-    def age(self):  # can fail if called before cache is populated and path doesn't exists
-        if self.path is None: return datetime.now() - self.time_produced
-        # if not self.path.exists(): return
+    def age(self):
+        """Throws AttributeError if called before cache is populated and path doesn't exists"""
+        if self.path is None:  # memory-based cache.
+            return datetime.now() - self.time_produced
         return datetime.now() - datetime.fromtimestamp(self.path.stat().st_mtime)
     def __setstate__(self, state: dict[str, Any]) -> None: self.__dict__.update(state); self.path = P.home() / self.path if self.path is not None else self.path
     def __getstate__(self) -> dict[str, Any]: state = self.__dict__.copy(); state["path"] = self.path.rel2home() if self.path is not None else state["path"]; return state  # With this implementation, instances can be pickled and loaded up in different machine and still works.
@@ -651,9 +652,10 @@ def __call__(self, fresh: bool = False) -> 'T':  # type: ignore
                 age = datetime.now() - datetime.fromtimestamp(self.path.stat().st_mtime)
                 if self.logger: self.logger(f"⚠️ {self.name} cache: Reading cached values from `{self.path}`. Lag = {age} ...")
                 self.cache = self.reader(self.path)
+                return self(fresh=False)  # may be the cache is old ==> check that by passing it through the logic again.
             else:
                 if self.logger: self.logger(f"⚠️ {self.name} cache: Populating fresh cache from {self.source_func}. Previous cache never existed or a there was an explicit fresh order.")
-                self.cache = self.source_func()
+                self.cache = self.source_func()  # fresh data.
                 if self.path is None: self.time_produced = datetime.now()
                 else: self.save(self.cache, self.path)
         else:  # cache exists

myresources/crocodile/meta.py

@@ -169,17 +169,18 @@ def run(self, *cmds: str, shell: Optional[SHELLS] = None, check: bool = False, i
         return Response.from_completed_process(resp)
     def run_script(self, script: str, shell: SHELLS = "default", verbose: bool = False):
         if self.machine == "Linux": script = "#!/bin/bash" + "\n" + script  # `source` is only available in bash.
-        tmp_file = P.tmpfile(name="tmp_shell_script", suffix=".ps1" if self.machine == "Windows" else ".sh", folder="tmp_scripts").write_text(script, newline={"Windows": None, "Linux": "\n"}[self.machine])
+        script_file = P.tmpfile(name="tmp_shell_script", suffix=".ps1" if self.machine == "Windows" else ".sh", folder="tmp_scripts").write_text(script, newline={"Windows": None, "Linux": "\n"}[self.machine])
         if shell == "default":
             if self.machine == "Windows":
                 start_cmd = "powershell"  # default shell on Windows is cmd which is not very useful. (./source is not available)
-                full_command: Union[list[str], str] = [start_cmd, str(tmp_file)]
+                full_command: Union[list[str], str] = [start_cmd, str(script_file)]  # shell=True will cause this to be a string anyway (with space separation)
             else:
-                start_cmd  = "."
-                full_command = f"{start_cmd} {tmp_file}"
+                start_cmd  = "bash"
+                full_command = f"{start_cmd} {script_file}"
+                # full_command = [start_cmd, str(script_file)]
         else:
-            full_command = [shell, str(tmp_file)]
-
+            # full_command = [shell, str(tmp_file)]
+            full_command = f"{shell} {script_file}"
         if verbose:
             from machineconfig.utils.utils import print_code
             print_code(code=script, lexer="shell", desc="Script to be executed:")