Merge pull request #41 from ni/v0.1.10_get_remove_entries_from_data_pane

v0.1.10: Get and remove entries from the data files pane

examples/Basic/run_test_session_and_show_log_file.py

@@ -0,0 +1,32 @@
+import os
+import sys
+
+from flexlogger.automation import Application, LogFileType
+
+
+def main(project_path):
+    """Launch FlexLogger, open a project, run a test session, and show log file."""
+    with Application.launch() as app:
+        project = app.open_project(path=project_path)
+        logging_specification = project.open_logging_specification_document()
+        logging_specification.remove_log_files(delete_files=True)
+        test_session = project.test_session
+        test_session.start()
+        print("Test started. Press Enter to stop the test and close the project...")
+        input()
+        test_session.stop()
+        log_files = logging_specification.get_log_files(LogFileType.TDMS)
+        project.close()
+        print("The following TDMS log files were created during the test session:")
+        for log_file in log_files:
+            print(log_file)
+    return 0
+
+
+if __name__ == "__main__":
+    argv = sys.argv
+    if len(argv) < 2:
+        print("Usage: %s <path of project to open>" % os.path.basename(__file__))
+        sys.exit()
+    project_path_arg = argv[1]
+    sys.exit(main(project_path_arg))

protobuf/ConfigurationBasedSoftware/FlexLogger/Automation/FlexLogger.Automation.Protocols/LoggingSpecificationDocument.proto

@@ -24,6 +24,10 @@ service LoggingSpecificationDocument {
     rpc GetLogFileDescription(GetLogFileDescriptionRequest) returns (GetLogFileDescriptionResponse) {}
     // RPC call to set the description
     rpc SetLogFileDescription(SetLogFileDescriptionRequest) returns (google.protobuf.Empty) {}
+    // RPC call to get items from the data files pane
+    rpc GetLogFiles(GetLogFilesRequest) returns (GetLogFilesResponse) {}
+    // RPC call to clear the data files pane
+    rpc RemoveLogFiles(RemoveLogFilesRequest) returns (google.protobuf.Empty) {}
     // RPC call to get all test properties
     rpc GetTestProperties(GetTestPropertiesRequest) returns (GetTestPropertiesResponse) {}
     // RPC call to set all test properties
@@ -128,6 +132,38 @@ message SetLogFileDescriptionRequest {
     string log_file_description = 2;
 }
 
+// Log file types
+enum LogFileType {
+    // TDMS files
+    TDMS = 0;
+    // CSV files
+    CSV = 1;
+    // TDMS backup files
+    TDMS_BACKUP = 2;
+}
+
+// Request object for GetLogFiles
+message GetLogFilesRequest {
+    // The id for the logging specification document
+    national_instruments.diagram_sdk.automation.protocols.ElementIdentifier document_identifier = 1;
+    // The type of log files to get
+    LogFileType log_file_type = 2;
+}
+
+// Response object for GetLogFiles
+message GetLogFilesResponse {
+    // The full paths on disk of the log files, sorted chronologically by file creation time
+    repeated string log_files = 1;
+}
+
+// Request object for RemoveLogFiles
+message RemoveLogFilesRequest {
+    // The id for the logging specification document
+    national_instruments.diagram_sdk.automation.protocols.ElementIdentifier document_identifier = 1;
+    // Delete files on disk?
+    bool delete_files = 2;
+}
+
 // Message that defines an individual test property
 message TestProperty {
     string property_name = 1;

setup.py

@@ -56,7 +56,7 @@ def _get_version(name: str) -> str:
     script_dir = os.path.dirname(os.path.realpath(__file__))
     script_dir = os.path.join(script_dir, name)
     if not os.path.exists(os.path.join(script_dir, "VERSION")):
-        version = "0.1.9"
+        version = "0.1.10"
     else:
         with open(os.path.join(script_dir, "VERSION"), "r") as version_file:
             version = version_file.read().rstrip()

src/flexlogger/automation/__init__.py

@@ -5,6 +5,7 @@
 from ._test_session_state import TestSessionState
 from ._channel_specification_document import ChannelSpecificationDocument
 from ._logging_specification_document import LoggingSpecificationDocument
+from ._log_file_type import LogFileType
 from ._screen_document import ScreenDocument
 from ._test_specification_document import TestSpecificationDocument
 from ._flexlogger_error import FlexLoggerError

src/flexlogger/automation/_flexlogger_error.py

@@ -19,18 +19,33 @@ def __init__(self, message: str) -> None:
     @property
     def message(self) -> str:
         """The error message."""
-        message = self._message
-        if isinstance(self.__cause__, RpcError):
-            cause = cast(RpcError, self.__cause__)
-            search_result = re.search(r"\(([^)]+)", cause.details())
-            inner_details = search_result.group(1)  # type: ignore
-            if len(inner_details) >= 0:
-                message += ". Additional error details: " + inner_details
-
-        return message
+        inner_details = self._get_inner_details()
+        if len(inner_details) == 0:
+            return self._message
+
+        inner_details = re.sub(r"\[([0-9+-]+)\] ", "", inner_details)
+        return self._message + ". Additional error details: " + inner_details
+
+    @property
+    def error_code(self) -> int:
+        """The error code."""
+        inner_details = self._get_inner_details()
+        if len(inner_details) == 0:
+            return 0
+
+        error_code_result = re.search(r"\[([0-9+-]+)\] ", inner_details)
+        return int(error_code_result.group(1)) if error_code_result else 0
 
     def __repr__(self) -> str:
         return f"FlexLoggerError({repr(self.message)})"
 
     def __str__(self) -> str:
         return self.message
+
+    def _get_inner_details(self) -> str:
+        if not isinstance(self.__cause__, RpcError):
+            return ""
+
+        cause = cast(RpcError, self.__cause__)
+        search_result = re.search(r"\(([^)]+)", cause.details())
+        return search_result.group(1)  # type: ignore

src/flexlogger/automation/_log_file_type.py

@@ -0,0 +1,13 @@
+from enum import Enum
+
+class LogFileType(Enum):
+    """An enumeration describing the different log file types."""
+
+    TDMS = 0
+    """TDMS files"""
+
+    TDMS_BACKUP_FILES = 1
+    """TDMS backup files"""
+
+    CSV = 2
+    """CSV files"""

src/flexlogger/automation/_logging_specification_document.py

@@ -1,6 +1,7 @@
 from google.protobuf.duration_pb2 import Duration
 from google.protobuf.timestamp_pb2 import Timestamp
 import datetime
+from datetime import timezone
 from dateutil import parser
 from dateutil import tz
 from grpc import Channel, RpcError
@@ -10,10 +11,18 @@
 from ._start_trigger_condition import StartTriggerCondition
 from ._stop_trigger_condition import StopTriggerCondition
 from ._test_property import TestProperty
+from ._log_file_type import LogFileType
 from ._value_change_condition import ValueChangeCondition
 from .proto import LoggingSpecificationDocument_pb2, LoggingSpecificationDocument_pb2_grpc
 from .proto.Identifiers_pb2 import ElementIdentifier
 
+from .proto.LoggingSpecificationDocument_pb2 import LogFileType as LogFileType_pb2
+
+LOG_FILE_TYPE_MAP = {
+    LogFileType.TDMS: LogFileType_pb2.TDMS,
+    LogFileType.CSV: LogFileType_pb2.CSV,
+    LogFileType.TDMS_BACKUP_FILES: LogFileType_pb2.TDMS_BACKUP,
+}
 
 class LoggingSpecificationDocument:
     """Represents a document that describes how data is logged.
@@ -205,6 +214,53 @@ def set_log_file_description(self, log_file_description: str) -> None:
             self._raise_if_application_closed()
             raise FlexLoggerError("Failed to set log file description") from error
 
+    def get_log_files(self, log_file_type: LogFileType) -> List[str]:
+        """Get log files in the data files pane of the project.
+
+        Args:
+            log_file_type: The type of log files to get.
+
+        Returns:
+            A list of the log files in the project.
+            The entries are sorted chronologically with the most recent file last.
+
+        Raises:
+            FlexLoggerError: if getting the log files fails.
+        """
+        stub = LoggingSpecificationDocument_pb2_grpc.LoggingSpecificationDocumentStub(self._channel)
+        try:
+            response = stub.GetLogFiles(
+                LoggingSpecificationDocument_pb2.GetLogFilesRequest(
+                    document_identifier=self._identifier,
+                    log_file_type = LOG_FILE_TYPE_MAP[log_file_type]
+                )
+            )
+            return [log_file for log_file in response.log_files]
+        except (RpcError, ValueError) as error:
+            self._raise_if_application_closed()
+            raise FlexLoggerError("Failed to get data files") from error
+
+    def remove_log_files(self, delete_files: bool = False) -> None:
+        """Remove log files from the data files pane of the project.
+
+        Args:
+            delete_files: True to delete files on disk, False to remove only from project.
+
+        Raises:
+            FlexLoggerError: if removing the log files fails.
+        """
+        stub = LoggingSpecificationDocument_pb2_grpc.LoggingSpecificationDocumentStub(self._channel)
+        try:
+            stub.RemoveLogFiles(
+                LoggingSpecificationDocument_pb2.RemoveLogFilesRequest(
+                    document_identifier=self._identifier,
+                    delete_files=delete_files
+                )
+            )
+        except (RpcError, ValueError) as error:
+            self._raise_if_application_closed()
+            raise FlexLoggerError("Failed to remove log files") from error
+
     def _convert_to_test_property(
         self, test_property: LoggingSpecificationDocument_pb2.TestProperty
     ) -> TestProperty:
@@ -392,7 +448,7 @@ def get_start_trigger_settings(self):
             self._raise_if_application_closed()
             raise FlexLoggerError("Failed to get the start trigger settings") from error
 
-    def get_stop_trigger_settings(self) -> (StopTriggerCondition, str):
+    def get_stop_trigger_settings(self) -> tuple[StopTriggerCondition, str]:
         """Get the stop trigger settings.
 
         Returns:

tests/test_logging_specification_document.py

@@ -10,7 +10,9 @@
 from flexlogger.automation import (
     Application,
     FlexLoggerError,
+    LogFileType,
     LoggingSpecificationDocument,
+    Project,
     StartTriggerCondition,
     StopTriggerCondition,
     TestProperty,
@@ -19,7 +21,7 @@
 )
 from nptdms import TdmsFile  # type: ignore
 
-from .utils import get_project_path, open_project
+from .utils import get_project_path, open_project, copy_project
 
 
 @pytest.fixture(scope="class")
@@ -32,6 +34,21 @@ def logging_spec_with_test_properties(app: Application) -> Iterator[LoggingSpeci
     with open_project(app, "ProjectWithTestProperties") as project:
         yield project.open_logging_specification_document()
 
+@pytest.fixture(scope="class")
+def project_with_produced_data(app: Application) -> Iterator[Project]:
+    """Fixture for opening ProjectWithProducedData.
+
+    This is useful to improve test time by not opening/closing this project in every test.
+    """
+    with copy_project("ProjectWithProducedData") as project_path:
+        project = app.open_project(project_path)
+        yield project
+        try:
+            project.close()
+        except FlexLoggerError:
+            # utils.kill_all_open_flexloggers may have killed this process already, that's fine
+            pass
+
 
 class TestLoggingSpecificationDocument:
     @pytest.mark.integration  # type: ignore
@@ -135,6 +152,59 @@ def test__open_project__set_logging_description__logging_path_updates(self, app:
 
             assert new_description == logging_specification.get_log_file_description()
 
+
+    @pytest.mark.integration  # type: ignore
+    def test__test_session_ran__remove_log_files__no_log_file_returned(
+        self, app: Application, project_with_produced_data: Project
+    ) -> None:
+        project = project_with_produced_data
+        project.test_session.start()
+        sleep(2.0)
+        project.test_session.stop()
+
+        logging_specification = project.open_logging_specification_document()
+        logging_specification.remove_log_files(delete_files=True)
+
+        log_files = logging_specification.get_log_files(LogFileType.TDMS)
+        assert len(log_files) == 0
+
+    @pytest.mark.integration  # type: ignore
+    def test__test_session_ran__get_log_files__log_file_returned(
+        self, app: Application, project_with_produced_data: Project
+    ) -> None:
+        project = project_with_produced_data
+        logging_specification = project.open_logging_specification_document()
+        logging_specification.remove_log_files(delete_files=True)
+        project.test_session.start()
+        sleep(2.0)
+        project.test_session.stop()
+
+        log_files = logging_specification.get_log_files(LogFileType.TDMS)
+
+        assert len(log_files) == 1
+        assert Path(log_files[0]).exists() is True
+
+    @pytest.mark.integration  # type: ignore
+    def test__test_session_ran_twice__get_log_files__two_log_files_returned(
+        self, app: Application, project_with_produced_data: Project
+    ) -> None:
+        project = project_with_produced_data
+        logging_specification = project.open_logging_specification_document()
+        logging_specification.remove_log_files(delete_files=True)
+        project.test_session.start()
+        sleep(2.0)
+        project.test_session.stop()
+        project.test_session.start()
+        sleep(2.0)
+        project.test_session.stop()
+
+        log_files = logging_specification.get_log_files(LogFileType.TDMS)
+
+        assert len(log_files) == 2
+        assert Path(log_files[0]).exists() is True
+        assert Path(log_files[1]).exists() is True
+        assert Path(log_files[0]).stat().st_ctime < Path(log_files[1]).stat().st_ctime
+
     @pytest.mark.integration  # type: ignore
     def test__open_project__get_test_properties__all_properties_returned(
         self, app: Application, logging_spec_with_test_properties: LoggingSpecificationDocument