adding conversions to the function list

Author: miranov25

UTILS/dfextensions/AliasDataFrame.py

@@ -112,6 +112,12 @@ def _default_functions(self):
         env["np"] = np
         for sf_name, sf_entry in self._subframes.items():
             env[sf_name] = sf_entry['frame']
+            # Custom compatibility for SetAlias-like expressions
+        env["int"] = lambda x: np.array(x).astype(np.int32)
+        env["uint"] = lambda x: np.array(x).astype(np.uint32)
+        env["float"] = lambda x: np.array(x).astype(np.float32)
+        env["round"] = np.round
+        env["clip"] = np.clip
         return env
 
     def _prepare_subframe_joins(self, expr):

UTILS/perfmonitor/test_performance_logger.py

@@ -1,53 +1,93 @@
-# test_performance_logger.py
-import unittest
+import time
+import tempfile
 import os
+import pytest
 import pandas as pd
-from UTILS.perfmonitor.performance_logger import PerformanceLogger, default_plot_config, default_summary_config
-
-class TestPerformanceLogger(unittest.TestCase):
-    def setUp(self):
-        self.log_path = "test_log.txt"
-        self.logger = PerformanceLogger(self.log_path)
-        # Ensure the log file is empty before each test
-        if os.path.exists(self.log_path):
-            os.remove(self.log_path)
-
-    def tearDown(self):
-        # Clean up the log file after each test
-        if os.path.exists(self.log_path):
-            os.remove(self.log_path)
-
-    def test_log(self):
-        self.logger.log("TestStep")
-        self.assertTrue(os.path.exists(self.log_path))
-        with open(self.log_path, "r") as f:
-            lines = f.readlines()
-        self.assertEqual(len(lines), 1)
-        self.assertIn("TestStep", lines[0])
-
-    def test_log_to_dataframe(self):
-        self.logger.log("TestStep")
-        df = PerformanceLogger.log_to_dataframe(self.log_path)
-        self.assertEqual(len(df), 1)
-        self.assertEqual(df.iloc[0]["step"], "TestStep")
-
-    def test_summarize_with_config(self):
-        self.logger.log("Step1")
-        self.logger.log("Step2")
-        df = PerformanceLogger.log_to_dataframe(self.log_path)
-        summary = PerformanceLogger.summarize_with_config(df, default_summary_config["summary_by_step"])
-        self.assertIn("elapsed_sec", summary.columns)
-        self.assertIn("rss_gb", summary.columns)
-
-    def test_plot(self):
-        self.logger.log("Step1")
-        self.logger.log("Step2")
-        df = PerformanceLogger.log_to_dataframe(self.log_path)
-        try:
-            PerformanceLogger.plot(df, default_plot_config)
-        except Exception as e:
-            self.fail(f"Plotting failed with exception: {e}")
-
-if __name__ == "__main__":
-    unittest.main()
+from perfmonitor.performance_logger import (
+    PerformanceLogger,
+    default_summary_config,
+    default_plot_config,
+)
 
+def test_basic_logging_and_parsing():
+    with tempfile.NamedTemporaryFile(delete=False, mode='w+', suffix=".txt") as tmp:
+        log_path = tmp.name
+
+    logger = PerformanceLogger(log_path)
+    logger.log("start")
+    time.sleep(0.1)
+    logger.log("step::loop", index=[0])
+    time.sleep(0.1)
+    logger.log("step::loop", index=[1, 2])
+
+    df = PerformanceLogger.log_to_dataframe([log_path])
+    assert not df.empty
+    assert "step" in df.columns
+    assert "elapsed_sec" in df.columns
+    assert "rss_gb" in df.columns
+    assert df["step"].str.contains("step::loop").any()
+    assert "index_1" in df.columns  # tests index parsing
+
+    os.remove(log_path)
+
+
+def test_missing_log_file_handling():
+    df = PerformanceLogger.log_to_dataframe(["nonexistent_file.txt"])
+    assert isinstance(df, pd.DataFrame)
+    assert df.empty
+
+
+def test_plot_and_summary(tmp_path):
+    log_path = tmp_path / "log.txt"
+    logger = PerformanceLogger(log_path)
+    logger.log("init")
+    time.sleep(0.05)
+    for i in range(3):
+        logger.log("step::loop", index=[i])
+        time.sleep(0.01)
+
+    df = PerformanceLogger.log_to_dataframe([str(log_path)])
+
+    summary = PerformanceLogger.summarize_with_config(df, default_summary_config)
+    assert isinstance(summary, dict)
+    assert "summary_by_step" in summary
+
+    # Test plotting (non-crashing)
+    PerformanceLogger.plot(df, default_plot_config)
+
+
+def test_multiple_files():
+    paths = []
+    for i in range(2):
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as tmp:
+            path = tmp.name
+        logger = PerformanceLogger(path)
+        logger.log(f"file{i}::start")
+        paths.append(path)
+
+    df = PerformanceLogger.log_to_dataframe(paths)
+    assert len(df) == 2
+    assert "logfile" in df.columns
+    for path in paths:
+        os.remove(path)
+
+
+def test_custom_summary():
+    with tempfile.NamedTemporaryFile(delete=False) as tmp:
+        log_path = tmp.name
+
+    logger = PerformanceLogger(log_path)
+    for i in range(3):
+        logger.log("step::measure", index=[i])
+        time.sleep(0.01)
+
+    df = PerformanceLogger.log_to_dataframe([log_path])
+    config = {
+        "by_index": {
+            "by": ["index_0"],
+            "stats": ["mean", "count"]
+        }
+    }
+    summary = PerformanceLogger.summarize_with_config(df, config)
+    assert "by_index" in summary
+    os.remove(log_path)