Merge pull request #15 from PTCInc/Add-Save-Load-Services

Add ProjectSave and ProjectLoad services

Author: rlabbeptc

kepconfig/connection.py

@@ -157,6 +157,10 @@ def SSL_trust_all_certs(self, val):
     def reinitialize(self, job_ttl = None) -> KepServiceResponse:
         '''Executes a Reinitialize call to the Kepware instance.
 
+        INPUTS:
+
+        "job_ttl" (optional) - Determines the number of seconds a job instance will exist following completion.
+
         RETURNS:
         KepServiceResponse instance with job information
 
@@ -167,7 +171,7 @@ def reinitialize(self, job_ttl = None) -> KepServiceResponse:
         '''
         url = self.url + self.__project_services_url + '/ReinitializeRuntime' 
         try:
-            job = self._kep_service_execute(url,job_ttl)
+            job = self._kep_service_execute(url, None, job_ttl)
             return job
         except Exception as err:
             raise err
@@ -236,6 +240,74 @@ def modify_project_properties(self, DATA, force = False) -> bool:
         if r.code == 200: return True 
         else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
 
+    def save_project(self, filename: str, password: str = None, job_ttl: int = None) -> KepServiceResponse:
+        '''Executes a ProjectSave Service call to the Kepware instance. This saves 
+        a copy of the current project file to disk. The filename
+
+        INPUTS:
+        
+        "filename" - Fully qualified relative file path and name of project file including the file extension.
+        location of project file save defaults: 
+                TKS or KEP (Windows): C:\\PROGRAMDATA\\PTC\\Thingworx Kepware Server\\V6 or 
+                                C:\\PROGRAMDATA\\Kepware\\KEPServerEX\\V6
+                TKE (Linux):    /opt/tkedge/v1/user_data
+
+
+        "password" (optional) - Specify a password with which to load or save an encrypted project file.  
+            This password will be required to load this project file.
+        
+        "job_ttl" (optional) - Determines the number of seconds a job instance will exist following completion.
+
+        RETURNS:
+        KepServiceResponse instance with job information
+
+        EXCEPTIONS (If not HTTP 200 or 429 returned):
+        
+        KepHTTPError - If urllib provides an HTTPError
+        KepURLError - If urllib provides an URLError
+        '''
+        url = self.url + self.__project_services_url + '/ProjectSave'
+        prop_data = {'servermain.PROJECT_FILENAME': filename}
+        if password != None: prop_data['servermain.PROJECT_PASSWORD'] = password
+        try:
+            job = self._kep_service_execute(url, prop_data, job_ttl)
+            return job
+        except Exception as err:
+            raise err
+
+    def load_project(self, filename: str, password: str = None, job_ttl: int = None) -> KepServiceResponse:
+        '''Executes a ProjectLoad Service call to the Kepware instance. This loads 
+        a project file to disk.
+
+        INPUTS:
+        
+        "filename" - Fully qualified path and name of project file including the file extension. Absolute
+        paths required for TKS and KEP while file path is relative for TKE
+            ex: Windows - filename = C:\\filepath\\test.opf
+                Linux - filename = filepath/test.lpf - Location is /opt/tkedge/v1/user_data/filepath/test.lpf
+
+        "password" (optional) - Specify a password with which to load or save an encrypted project file.
+        
+        "job_ttl" (optional) - Determines the number of seconds a job instance will exist following completion.
+
+        RETURNS:
+        KepServiceResponse instance with job information
+
+        EXCEPTIONS (If not HTTP 200 or 429 returned):
+        
+        KepHTTPError - If urllib provides an HTTPError
+        KepURLError - If urllib provides an URLError
+        '''
+        url = self.url + self.__project_services_url + '/ProjectLoad'
+        prop_data = {'servermain.PROJECT_FILENAME': filename}
+        if password != None: prop_data['servermain.PROJECT_PASSWORD'] = password
+        try:
+            job = self._kep_service_execute(url, prop_data, job_ttl)
+            return job
+        except Exception as err:
+            raise err
+
+
     def service_status(self, resp: KepServiceResponse):
         '''Returns the status of a service job. Used to verify if a service call
         has completed or not.
@@ -325,11 +397,12 @@ def _force_update_check(self, force, DATA):
                         pass
         return DATA
     # General service call handler
-    def _kep_service_execute(self, url, TTL = None):
+    def _kep_service_execute(self, url, DATA = None, TTL = None):
         try:
             if TTL != None:
-                TTL = {"servermain.JOB_TIME_TO_LIVE_SECONDS": TTL}
-            r = self._config_update(url, TTL)
+                if DATA == None: DATA = {}
+                DATA["servermain.JOB_TIME_TO_LIVE_SECONDS"]= TTL
+            r = self._config_update(url, DATA)
             job = KepServiceResponse(r.payload['code'],r.payload['message'], r.payload['href'])
             return job
         except KepHTTPError as err:

kepconfig/connectivity/device.py

@@ -180,6 +180,8 @@ def auto_tag_gen(server, device_path, job_ttl = None) -> KepServiceResponse:
     "device_path" -  path identifying device. Standard Kepware address decimal notation string including the 
     device such as "channel1.device1"
 
+    "job_ttl" (optional) - Determines the number of seconds a job instance will exist following completion.
+
     RETURNS:
     KepServiceReturn with the result of the service either Code 202 (Accepted) or 429 (Too Busy)
 
@@ -191,7 +193,7 @@ def auto_tag_gen(server, device_path, job_ttl = None) -> KepServiceResponse:
     path_obj = kepconfig.path_split(device_path)
     try:
         url = server.url +channel._create_url(path_obj['channel']) + _create_url(path_obj['device']) + ATG_URL
-        job = server._kep_service_execute(url, job_ttl)
+        job = server._kep_service_execute(url, None, job_ttl)
         return job
     except KeyError as err:
         err_msg = 'Error: No {} identified in {} | Function: {}'.format(err,'device_path', inspect.currentframe().f_code.co_name)

kepconfig/datalogger/log_group.py

@@ -202,5 +202,5 @@ def reset_column_mapping_service(server, log_group, job_ttl = None) -> KepServic
     '''
 
     url = server.url + _create_url(log_group) + SERVICES_ROOT + '/ResetColumnMapping'
-    job = server._kep_service_execute(url, job_ttl)
+    job = server._kep_service_execute(url, None, job_ttl)
     return job

tests/admin_test.py

@@ -7,12 +7,13 @@
 # Admin Test - Test to exectute various calls for the Administrator 
 # parts of the Kepware configuration API
 
+from wsgiref.simple_server import server_version
 from kepconfig.error import KepError, KepHTTPError
 import os, sys
 from typing import Dict, List
 import pytest
 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-# import kepconfig
+import kepconfig
 from kepconfig import admin
 import json
 import time
@@ -28,6 +29,8 @@
 twx_agent_name = 'Thingworx'
 iot_item_name ="System__Date"
 
+
+
 group1 = {'common.ALLTYPES_NAME': 'Operators'}
 group2 = {'common.ALLTYPES_NAME': 'Group1'}
 group3 = {'common.ALLTYPES_NAME': 'Group2'}
@@ -89,8 +92,10 @@ def complete(server):
 
 
 @pytest.fixture(scope="module")
-def server(kepware_server):
-    server = kepware_server
+def server(kepware_server: list[kepconfig.connection.server, str]):
+    server = kepware_server[0]
+    global server_type
+    server_type = kepware_server[1]
 
     # Initialize any configuration before testing in module
     initialize(server)
@@ -100,10 +105,7 @@ def server(kepware_server):
     complete(server)
 
 def test_uaserver(server):
-    try:
-        server._config_get(server.url + admin.ua_server._create_url())
-    except Exception as err:
-        pytest.skip("UA Endpoints not configurable.")
+    if server_type == 'TKS': pytest.skip("UA Endpoints not configurable in {}.".format(server_type))
 
     assert admin.ua_server.add_endpoint(server,uaendpoint1)
 

tests/conftest.py

@@ -11,5 +11,8 @@
 
 @pytest.fixture(scope="module")
 def kepware_server():
-    return kepconfig.connection.server(host = 'localhost', port = 57412, user = 'Administrator', pw = '', https = False)
-    # return kepconfig.connection.server(host = 'localhost', port = 57413, user = 'Administrator', pw = 'Kepware400400400', https = False)
+    # return [kepconfig.connection.server(host = 'localhost', port = 57412, user = 'Administrator', pw = '', https = False), 'TKS']
+    
+    server = kepconfig.connection.server(host = '127.0.0.1', port = 57513, user = 'Administrator', pw = 'Kepware400400400', https = True)
+    server.SSL_trust_all_certs = True
+    return [server, 'TKE']

tests/connectivity_test.py

@@ -59,8 +59,10 @@ def complete(server):
         HTTPErrorHandler(err)
 
 @pytest.fixture(scope="module")
-def server(kepware_server):
-    server = kepware_server
+def server(kepware_server: list[kepconfig.connection.server, str]):
+    server = kepware_server[0]
+    global server_type
+    server_type = kepware_server[1]
 
     # Initialize any configuration before testing in module
     initialize(server)

tests/datalogger_test.py

@@ -155,6 +155,8 @@ def HTTPErrorHandler(err):
         print('Different Exception Received: {}'.format(err))
 
 def initialize(server: kepconfig.connection.server):
+	if server_type == 'TKE': pytest.skip("Datalogger not configurable in {}.".format(server_type), allow_module_level=True)
+	
 	try:
 		server._config_get(server.url + datalogger.log_group._create_url())
 	except Exception as err:
@@ -169,8 +171,10 @@ def complete(server):
 		HTTPErrorHandler(err)	
 
 @pytest.fixture(scope="module")
-def server(kepware_server):
-    server = kepware_server
+def server(kepware_server: list[kepconfig.connection.server, str]):
+    server = kepware_server[0]
+    global server_type
+    server_type = kepware_server[1]
 
     # Initialize any configuration before testing in module
     initialize(server)

tests/egd_test.py

@@ -118,6 +118,8 @@ def HTTPErrorHandler(err):
 }
 
 def initialize(server: kepconfig.connection.server):
+    if server_type == 'TKE': pytest.skip("EGD not configurable in {}.".format(server_type), allow_module_level=True)
+    
     try:
         server._config_get(server.url +"/doc/drivers/GE Ethernet Global Data/channels")
     except Exception as err:
@@ -135,8 +137,10 @@ def complete(server):
             HTTPErrorHandler(err)
 
 @pytest.fixture(scope="module")
-def server(kepware_server):
-    server = kepware_server
+def server(kepware_server: list[kepconfig.connection.server, str]):
+    server = kepware_server[0]
+    global server_type
+    server_type = kepware_server[1]
 
     # Initialize any configuration before testing in module
     initialize(server)

tests/iot_gateway_test.py

@@ -91,8 +91,10 @@ def complete(server):
             HTTPErrorHandler(err)
 
 @pytest.fixture(scope="module")
-def server(kepware_server):
-    server = kepware_server
+def server(kepware_server: list[kepconfig.connection.server, str]):
+    server = kepware_server[0]
+    global server_type
+    server_type = kepware_server[1]
 
     # Initialize any configuration before testing in module
     initialize(server)