Added code: scx00.py

README.md

@@ -1,7 +1,5 @@
 # python-eaton-scx00
 
-**I am goingto add the code soon.**
-
 ## Overview
 
 SCX00.py is a module for reading and writing data from/to Eaton Power SC200 and SC300 Controllers.

scx00.py

@@ -0,0 +1,356 @@
+
+"""SCx00.py 
+
+   Python module for reading and writing data from/to Eaton Power SC200 and SC300 Controllers.
+   I don't know whether it also works with the SC100 model as I don't have access to one.
+
+   The module implements a simple wrapper for the Eaton SCx00 XMLRPC interface. The SCx00 class
+   provides dictionary style access to all data items.
+
+   No error checking is done and all XMLRPC errors are passed on to the caller.
+
+   Status: Works for me ;-)
+
+   Markus Juenemann
+
+   Big thanks to 'The-Godfather' for having figured out how to make xmlrpc calls with
+   cookies and 'GermainZ' for having asked the question on Stackoverflow. 
+   The (Safe)CookieTransport implementations below are based on their code.
+
+"""
+
+__author__ = 'Markus Juenemann <[email protected]>'
+__version__ = '0.1.0'
+__license__ = 'BSD 2-Clause'
+
+
+# Things that are needed somewhere else.
+#
+import urllib.parse
+import xmlrpc.client
+import ssl
+
+DEBUG = False
+TIMEOUT = 10
+
+
+# The Eaton SCx00 Controller wants authenication credentials sent as cookies.
+# As the default transports do not support this we have to create custom ones.
+#
+class SafeCookieTransport(xmlrpc.client.SafeTransport):
+    def __init__(self, context=None, username='', password='', **kwargs):
+        self.username = username
+        self.password = password
+        super().__init__(context=context, **kwargs)
+
+    def send_headers(self, connection ,headers):
+        cookie = 'delimiter=%2C'
+        if self.username and self.password:
+            cookie  = f'name={self.username}; pwd={self.password}; ' + cookie
+        connection.putheader('Cookie', cookie)
+        connection.timeout = TIMEOUT
+        super().send_headers(connection, headers)
+
+
+class CookieTransport(xmlrpc.client.Transport):
+    def __init__(self, username='', password='', **kwargs):
+        self.username = username
+        self.password = password
+        super().__init__(**kwargs)
+
+    def send_headers(self, connection ,headers):
+        cookie = 'delimiter=%2C'
+        if self.username and self.password:
+            cookie  = f'name={self.username}; pwd={self.password}; ' + cookie
+        connection.putheader('Cookie', cookie)
+        connection.timeout = TIMEOUT
+        super().send_headers(connection, headers)
+
+
+
+class SCx00:
+    """Represents a connection to an Eaton Powerware SCx00 Controller.
+
+       >>> client = SCx00('10.1.2.3')
+
+       Instances provides dictionary-style access to the data items of the conroller.
+
+       >>> client['Site-Name']
+       Site-12345
+
+       Write access probbaly requires to authenticate first. Either provide username
+       and password as keywork arguments when creating an instance or call the 
+       ``SCx00.login()`` method.
+
+       >>> client.login('myusername', 'mypassword')
+
+       >>> client['Site-Name']
+       Site-12345
+
+       >>> client['Site-Name'] = 'Site-98765'
+       >>> client['Site-Name']
+       Site-98765
+
+       Returned values may have to be cast into their appropriate type. Converting
+       an integer into an IPv4 address actually works as IPv4 addresses area really
+       just 32 bit integers.
+
+       >>> client['IP-Address']
+       167838211
+
+       >>> import ipaddress
+       >>> ipaddress.IPv4Address(client['IP-Address'])
+       IPv4Address('10.1.2.3')
+
+       Enumerations are returned as their integer value (1=Disable, 1=All, 2=Read Only, 3=V3 Only).
+
+       >>> client['SNMP-Access']
+       3
+
+       Boolean values are enumerations (0=Disabled, 1=Enabled) which fortunately can be directly
+       converted into the proper Python type and value.
+
+       >>> client['HTTPS-Access']
+       1
+       >>> bool(client['HTTPS-Access'])
+       True
+
+       >>> client['HTTP-Access']
+       0
+       >>> bool(client['HTTP-Access'])
+       False
+
+       Date/time values are easy to convert, too.
+
+       >>> client['Next-Battery-Test-Start-Time']
+       1639261885
+
+       >>> import datetime
+       >>> datetime.datetime.fromtimestamp(client['Next-Battery-Test-Start-Time'])
+       datetime.datetime(2021, 12, 12, 9, 31, 25)
+
+       Tabular can be accessed through a special slicing notation.
+
+       >>> client['DI-Name:1-8']
+       ['Surge Diverter Fault', 'Surge Filter Fault', 'UPS Fault', 'Battery Isolated',
+        'Panel Temp High', 'Door Alarm', 'UPS Mains Power Fail', 'Battery Fuse Fail']
+
+       It is up to the user to provide valid values for the lower and upper bounds of
+       a slice as these are passed varbatim to the controller's XMLRPC interface.
+       Please note that some  tables index data starting with 0, others with 1. For
+       example, 'DI-Name' is indexed starting with 1.
+
+       >>> client['DI-Name:0-8']
+       Traceback (most recent call last):
+         File "scx00.py", line 331, in <module>
+           print(client['DI-Name:0-10'])
+         File "scx00.py", line 291, in __getitem__
+           raise KeyError(f"{item} -> {resp}")
+       KeyError: "DI-Name:0-10 -> [[{'faultCode': 4, 'faultString': 'Exception: Index out of range on access to item DI-Name'}]]"
+
+       Individual entries can be accessed directly which is the only way to update an
+       item in a table.
+
+       >>> client['DI-Name:6']
+       Door Alarm
+
+       >>> client['DI-Name:6'] = 'Door Open Alarm'
+       >>> client['DI-Name:6']
+       Door Open Alarm
+       
+       Unknown items will raise a KeyError exception.
+
+       >>> client['No-Such-Item']
+       Traceback (most recent call last):
+         File "scx00.py", line 334, in <module>
+           print(client['No-Such-Item'])
+         File "scx00.py", line 256, in __getitem__
+           raise KeyError(f"{item} -> {resp}")
+       KeyError: "No-Such-Item -> [{'faultCode': 3, 'faultString': 'Unknown Item'}]"
+
+
+       Arguments:
+       controller -- Hostname or IP Address of the controller.
+
+       Keyword Arguments:
+       username -- Username for authentication.
+       password -- Password for authentication.
+       port -- The port the controller listens on. Defaults to 443.
+       proto -- Either 'http' or 'https'. Defaults to 'https'.
+
+       If 'username' and 'password' are not given, communications with the
+       SCx000 controller will be unauthenticated. Depending on the 
+       configuration of the SCx00 controller this may be limited to
+       read-only access or no access at all.
+
+       """
+
+
+    def __init__(self, controller, proto=None, port=0, username='', password=''):
+        assert isinstance(port, int)
+        assert proto in ("http", "https", None)
+
+        self.controller = controller
+        self.login(username, password)
+
+        if port == 80 and proto is None:
+            self.port = port
+            self.proto = 'http'
+        elif port == 443 and proto is None:
+            self.port = port
+            self.proto = 'https'
+        elif proto == 'https' and port is 0:
+            self.port = 443
+            self.proto = proto
+        elif proto == 'http' and port is 0:
+            self.port = 80
+            self.proto = proto
+        else:
+            self.port = 443
+            self.proto = 'https'
+
+
+    def login(self, username, password):
+        """Store username and password for later(!) authentication."""
+        self.username = username
+        self.password = password
+
+
+    @property
+    def url(self):
+        return urllib.parse.urlunparse((self.proto, f"{self.controller}:{self.port}", "xmlrpc", None, None, None))
+
+
+    @property
+    def transport(self):
+        if self.proto == 'https':
+            return SafeCookieTransport(context=ssl._create_unverified_context(), username=self.username, password=self.password)
+        else:
+            return CookieTransport(username=self.username, password=self.password)
+
+
+    def __getitem__(self, item):
+        """Read data item(s) from the controller.
+
+        """
+
+        assert isinstance(item ,str)
+
+
+        with xmlrpc.client.ServerProxy(self.url, transport=self.transport, verbose=DEBUG) as proxy:
+
+            resp = proxy.db.get([item])
+            """The response will always be a list. Examples:
+
+               Single items:
+
+                 [171027852] 
+                 ['CTRL-12345']
+
+               List of items, possible with "repeat" markers.
+
+                 [['Door Alarm', 'Cabinet Fan Active', 'Surge Filter Alarm', 'Surge Diverter Alarm', 'Digital Input 5']]
+                 [[0, {'repeatLast': 4}]]
+
+               Single items or lists contain fault messages.
+
+                 [['Low Float', ..., 'Normal Charge', {'faultCode': 4, 'faultString': 'Exception: Index out of range on access to item Alarm-Name'}]]
+
+                 [{'faultCode': 3, 'faultString': 'Unknown Item'}]
+
+            """
+
+
+            # I don't know what an empty response could mean. Does the controller ever return
+            # an empty response?
+            #
+            if not resp:
+                raise KeyError(f"{item} -> {resp}")
+
+
+            # Ensure that the resopnse is a list.
+            #
+            if not isinstance(resp, list):
+                raise KeyError(f"{item} -> {resp}")
+
+
+            # Extract the first item from the list as it contains the value.
+            #
+            value = resp[0]
+
+
+            # The server returned an error if the first item in the list is a dictionary.
+            #
+            if isinstance(value, dict):
+                raise KeyError(f"{item} -> {resp}")
+
+            # Return elementary types
+            #
+            elif isinstance(value, str):
+                return value
+            elif isinstance(value, int):
+                return value
+            elif isinstance(value, float):
+                return value
+
+            # The result may be a list. If so, every lement of the list must be
+            # checked for fault messages and repeat markers.
+            #
+            elif isinstance(value, list):
+                value2 = []
+                last = None
+
+                for v in value:
+
+                    # Dictionaries are special messages returned by the controller but everything
+                    # else are regular values.
+                    #
+                    if not isinstance(v, dict):
+                        value2.append(v)
+                        last = v
+
+                    # Handle {'repeatLast': 58}
+                    #
+                    elif 'repeatLast' in v:
+                        value2 += [last] * v['repeatLast']
+
+                    # Any faults cause a KeyError.
+                    # {'faultCode': 4, 'faultString': 'Exception: Index out of range on access to item ...'}
+                    #
+                    elif 'faultCode' in v:
+                        raise KeyError(f"{item} -> {resp}")
+
+                    # Any other error.
+                    #
+                    else:
+                        raise KeyError(f"{item} -> {resp}")
+
+
+            # Return the result list if it has not been done yet.
+            #
+            return value2
+
+
+
+
+    def __setitem__(self, item, value):
+        with xmlrpc.client.ServerProxy(self.url, transport=self.transport, verbose=DEBUG) as proxy:
+            resp = proxy.db.set({item: value})
+            if not isinstance(resp, dict):
+                raise KeyError(f"{item} -> {resp}")
+            elif resp.get('resultString', '') != 'OK':
+                raise KeyError(f"{item} -> {resp}")
+
+
+    def get(self, item, default=None):
+        # WARNING: It is the caller's responsibility to set `default` correclt for tables:
+        #          client.get('Di-Name:1-10', [None]*10)
+        try:
+            return self[item[0]]
+        except KeyError:
+            return default
+
+
+# Aliases
+#
+SCX00 = SC200 = SC300 = SCx00