Add disable channel functionality

examples/Basic/disable_channel.py

@@ -0,0 +1,26 @@
+import os
+import sys
+
+from flexlogger.automation import Application
+
+
+def main(project_path):
+    """Launch FlexLogger, open a project, and disables a channel."""
+    with Application.launch() as app:
+        project = app.open_project(path=project_path)
+        channel_name = input("Enter the name of the channel to disable: ")
+        channel_specification = project.open_channel_specification_document()
+        channel_specification.set_channel_enabled(channel_name, False)
+        print("Channel disabled. Press Enter to close the project...")
+        input()
+        project.close()
+    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/ChannelSpecificationDocument.proto

@@ -13,6 +13,10 @@ service ChannelSpecificationDocument {
     rpc GetDoubleChannelValue(GetDoubleChannelValueRequest) returns (GetDoubleChannelValueResponse) {}
     // RPC call to set the value of a double channel
     rpc SetDoubleChannelValue(SetDoubleChannelValueRequest) returns (SetDoubleChannelValueResponse) {}
+    // RPC call to get a channel's enable state
+    rpc IsChannelEnabled(IsChannelEnabledRequest) returns (IsChannelEnabledResponse) {}
+    // RPC call to enable/disable channels
+    rpc SetChannelEnabled(SetChannelEnabledRequest) returns (SetChannelEnabledResponse) {}
 }
 
 // Request object for getting all channel names
@@ -55,4 +59,31 @@ message SetDoubleChannelValueRequest {
 
 // Response object for setting a channel value
 message SetDoubleChannelValueResponse {
-}
+}
+
+// Request object for getting a channel enable state
+message IsChannelEnabledRequest {
+    // The id for the channel specification document
+    national_instruments.diagram_sdk.automation.protocols.ElementIdentifier document_identifier = 1;
+    // The name of the channel to get the enable state for
+    string channel_name = 2;
+}
+
+// Response object for getting a channel enable state
+message IsChannelEnabledResponse {
+    bool channel_enabled = 1;
+}
+
+// Request object for setting a channel enable state
+message SetChannelEnabledRequest {
+    // The id for the channel specification document
+    national_instruments.diagram_sdk.automation.protocols.ElementIdentifier document_identifier = 1;
+    // The name of the channel to enable or disable
+    string channel_name = 2;
+    // The enable state to set
+    bool channel_enabled = 3;
+}
+
+// Response object for setting a channel enable state
+message SetChannelEnabledResponse {
+}

requirements.txt

@@ -1,6 +1,7 @@
 console-menu
 grpcio-tools
 grpcio
+nptdms
 prettytable
 psutil
 # This package only works correctly on Windows,

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.1"
+        version = "0.1.2"
     else:
         with open(os.path.join(script_dir, "VERSION"), "r") as version_file:
             version = version_file.read().rstrip()

src/flexlogger/automation/_channel_specification_document.py

@@ -29,6 +29,28 @@ def __init__(
         self._raise_if_application_closed = raise_if_application_closed
         self._identifier = identifier
 
+    def is_channel_enabled(self, channel_name: str) -> bool:
+        """Get the current enabled state of the specified channel.
+
+        Args:
+            channel_name: The name of the channel.
+
+        Raises:
+            FlexLoggerError: if getting the channel value fails.
+        """
+        stub = ChannelSpecificationDocument_pb2_grpc.ChannelSpecificationDocumentStub(self._channel)
+        try:
+            response = stub.IsChannelEnabled(
+                ChannelSpecificationDocument_pb2.IsChannelEnabledRequest(
+                    document_identifier=self._identifier, channel_name=channel_name
+                )
+            )
+
+            return response.channel_enabled
+        except (RpcError, ValueError) as error:
+            self._raise_if_application_closed()
+            raise FlexLoggerError("Failed to get channel enable state") from error
+
     def get_channel_names(self) -> List[str]:
         """Get all the channel names in the document.
 
@@ -73,6 +95,29 @@ def get_channel_value(self, channel_name: str) -> ChannelDataPoint:
             self._raise_if_application_closed()
             raise FlexLoggerError("Failed to get channel value") from error
 
+    def set_channel_enabled(self, channel_name: str, channel_enabled: bool) -> None:
+        """Enable or disable the specified channel.
+
+        Args:
+            channel_name: The name of the channel.
+            channel_enabled: The channel enabled state: true to enable the channel, false to disable it.
+
+        Raises:
+            FlexLoggerError: if enabling or disabling the channel fails.
+        """
+        stub = ChannelSpecificationDocument_pb2_grpc.ChannelSpecificationDocumentStub(self._channel)
+        try:
+            stub.SetChannelEnabled(
+                ChannelSpecificationDocument_pb2.SetChannelEnabledRequest(
+                    document_identifier=self._identifier,
+                    channel_name=channel_name,
+                    channel_enabled=channel_enabled,
+                )
+            )
+        except (RpcError, ValueError) as error:
+            self._raise_if_application_closed()
+            raise FlexLoggerError("Failed to set the channel enable state") from error
+
     def set_channel_value(self, channel_name: str, channel_value: float) -> None:
         """Set the current value of the specified channel.
 

tests/test_channel_specification_document.py

@@ -109,3 +109,32 @@ def test__channeldatapoint_repr__returns_correct_string(self) -> None:
             channel_data_point.timestamp
         )
         assert expected_repr == repr(channel_data_point)
+
+    @pytest.mark.integration  # type: ignore
+    def test__project_with_writable_channels__disable_channel__channel_disabled(
+        self, app: Application
+    ) -> None:
+        with open_project(app, "ProjectWithSwitchboard") as project:
+            channel_specification = project.open_channel_specification_document()
+
+            channel_specification.set_channel_enabled("Switch 42", False)
+
+            channel_enabled = channel_specification.is_channel_enabled("Switch 42")
+
+            assert not channel_enabled
+
+    @pytest.mark.integration  # type: ignore
+    def test__set_channel_enabled_for_channel_that_does_not_exist__exception_raised(
+        self, app: Application, channels_with_produced_data: ChannelSpecificationDocument
+    ) -> None:
+        channel_specification = channels_with_produced_data
+        with pytest.raises(FlexLoggerError):
+            channel_specification.set_channel_enabled("Not a channel", True)
+
+    @pytest.mark.integration  # type: ignore
+    def test__set_channel_enabled_for_readonly_channel__exception_raised(
+        self, app: Application, channels_with_produced_data: ChannelSpecificationDocument
+    ) -> None:
+        channel_specification = channels_with_produced_data
+        with pytest.raises(FlexLoggerError):
+            channel_specification.set_channel_enabled("Channel 1", False)