fix filetime <> datetime convertion, crappy stuff

opcua/uatypes.py

@@ -2,7 +2,9 @@
 implement ua datatypes
 """
 from enum import Enum
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, tzinfo
+from calendar import timegm
+
 import uuid
 import struct 
 
@@ -11,18 +13,38 @@ import opcua.status_code as status_code
 #types that will packed and unpacked directly using struct (string, bytes and datetime are handles as special cases
 UaTypes = ("Boolean", "SByte", "Byte", "Int8", "UInt8", "Int16", "UInt16", "Int32", "UInt32", "Int64", "UInt64", "Float", "Double")
 
+
+EPOCH_AS_FILETIME = 116444736000000000  # January 1, 1970 as MS file time
+HUNDREDS_OF_NANOSECONDS = 10000000
+
+class UTC(tzinfo):
+    """UTC"""
+    def utcoffset(self, dt):
+        return timedelta(0)
+
+    def tzname(self, dt):
+        return "UTC"
+
+    def dst(self, dt):
+        return timedelta(0)
+
+
+#methods copied from  David Buxton <david@gasmark6.com> sample code
 def datetime_to_win_epoch(dt):
-    epch = (dt - datetime(1601, 1, 1, 0, 0)).total_seconds() * 10 ** 7
-    return int(epch)
+    if (dt.tzinfo is None) or (dt.tzinfo.utcoffset(dt) is None):
+        dt = dt.replace(tzinfo=UTC())
+    ft = EPOCH_AS_FILETIME + (timegm(dt.timetuple()) * HUNDREDS_OF_NANOSECONDS)
+    return ft + (dt.microsecond * 10)
 
 def win_epoch_to_datetime(epch):
-    return datetime(1601, 1, 1) + timedelta(microseconds=epch/10.0)
+    (s, ns100) = divmod(epch - EPOCH_AS_FILETIME, HUNDREDS_OF_NANOSECONDS)
+    dt = datetime.utcfromtimestamp(s)
+    dt = dt.replace(microsecond=(ns100 // 10))
+    return dt
+
+
 
 def uatype_to_fmt(uatype):
-    #if uatype == "String":
-        #return "s"
-    #elif uatype == "CharArray":
-        #return "s"
     if uatype == "Char":
         return "B"
     elif uatype == "SByte":

tests.py

 #! /usr/bin/env python
 import io
 import sys
-from datetime import datetime
+from datetime import datetime, timedelta
 import unittest
 from threading import Thread, Event
 try:
@@ -93,10 +93,18 @@ class Unit(unittest.TestCase):
         qn = ua.QualifiedName("Root", 0)
         v = Variant
 
+    def test_datetime(self):
+        now = datetime.now()
+        epch = ua.datetime_to_win_epoch(now)
+        dt = ua.win_epoch_to_datetime(epch)
+        self.assertEqual(now, dt)
 
-
-
-
+        epch = 128930364000001000
+        dt = ua.win_epoch_to_datetime(epch)
+        print("dt", dt)
+        epch2 = ua.datetime_to_win_epoch(dt)
+        print("epch2", epch2)
+        self.assertEqual(epch, epch2)
 
     def test_equal_nodeid(self):
         nid1 = ua.NodeId(999, 2)
@@ -145,7 +153,6 @@ class Unit(unittest.TestCase):
         self.assertEqual(dv.Value, ua.Variant('abc'))
         now = datetime.now() 
         dv.source_timestamp = now
-        #self.assertEqual(int(dv.source_timestamp.to_time_t()), tnow)
 
     def test_variant(self):
         dv = ua.Variant(True, ua.VariantType.Boolean)
@@ -156,7 +163,7 @@ class Unit(unittest.TestCase):
         self.assertEqual(v.Value, now)
         self.assertEqual(v.VariantType, ua.VariantType.DateTime)
         v2 = ua.Variant.from_binary(ua.utils.Buffer(v.to_binary()))
-        #self.assertEqual(v.Value, v2.Value)
+        self.assertEqual(v.Value, v2.Value)
         self.assertEqual(v.VariantType, v2.VariantType)
 
     def test_variant_array(self):
@@ -172,7 +179,7 @@ class Unit(unittest.TestCase):
         self.assertEqual(v.Value[0], now)
         self.assertEqual(v.VariantType, ua.VariantType.DateTime)
         v2 = ua.Variant.from_binary(ua.utils.Buffer(v.to_binary()))
-        #self.assertEqual(v.Value, v2.Value)
+        self.assertEqual(v.Value, v2.Value)
         self.assertEqual(v.VariantType, v2.VariantType)
 
 
@@ -268,10 +275,9 @@ class CommonTests(object):
     def test_datetime_read(self):
         time_node = self.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
         dt = time_node.get_value()
-        #pydt = dt.to_datetime()
-        #utcnow = datetime.datetime.utcnow()
-        #delta = utcnow - pydt
-        #self.assertTrue(delta < datetime.timedelta(seconds=1))
+        utcnow = datetime.utcnow()
+        delta = utcnow - dt
+        self.assertTrue(delta < timedelta(seconds=1))
 
     def test_datetime_write(self):
         time_node = self.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
@@ -279,7 +285,7 @@ class CommonTests(object):
         objects = self.opc.get_objects_node()
         v1 = objects.add_variable(4, "test_datetime", now)
         tid = v1.get_value()
-        #self.assertEqual(now, tid.to_datetime()) rounding error!!
+        self.assertEqual(now, tid) 
 
     def test_add_numeric_variable(self):
         objects = self.opc.get_objects_node()