Merge pull request #75 from pymeasure/fix-uuidv7

Rework UUIDv7 generation for conversation_id

environment.yml

@@ -3,10 +3,10 @@ channels:
   - conda-forge
 dependencies:
   - pyzmq  #=25.1.2  don't pin version for python<3.9
+  - uuid6=2024.1.12
   - pip  # don't pin, to gain newest conda compatibility fixes
-  - pip:
+  # - pip:
     # - openrpc==8.1.0  don't pin presence for python<3.9
-    - uuid7==0.1.0
 # Development dependencies below
   - pytest=7.2.0
   - pytest-cov=4.1.0

pyleco/core/serialization.py

@@ -23,11 +23,13 @@
 #
 
 from __future__ import annotations
+import datetime
 from enum import IntEnum, IntFlag
 import json
 from typing import Any, Optional, NamedTuple, Union
 
-from uuid_extensions import uuid7  # type: ignore  #  as long as uuid does not yet support UUIDv7
+# as long as uuid does not yet support UUIDv7 use uuid6
+from uuid6 import uuid7
 from ..json_utils.json_objects import (
     Request,
     ParamsRequest,
@@ -152,7 +154,12 @@ def deserialize_data(content: bytes) -> Any:
 
 def generate_conversation_id() -> bytes:
     """Generate a conversation_id."""
-    return uuid7(as_type="bytes")  # type: ignore
+    return uuid7().bytes
+
+
+def conversation_id_to_datetime(conversation_id: bytes) -> datetime.datetime:
+    seconds_since_epoch = int.from_bytes(conversation_id[:6], byteorder="big", signed=False) / 1000
+    return datetime.datetime.fromtimestamp(seconds_since_epoch, tz=datetime.timezone.utc)
 
 
 def _get_json_object_type(data: dict[str, Any]) -> JsonContentTypes:

pyproject.toml

@@ -22,7 +22,7 @@ requires-python = ">=3.8"
 dependencies = [
   "pyzmq >= 22.3.0",
   "openrpc >= 8.1.0; python_version >= '3.9'",
-  "uuid7 >= 0.1.0",
+  "uuid6 >= 2024.1.12",
 ]
 
 [project.optional-dependencies]

tests/core/test_serialization.py

@@ -23,6 +23,8 @@
 #
 
 from __future__ import annotations
+import datetime
+import uuid
 from typing import Any, Optional, Union
 
 import pytest
@@ -108,6 +110,23 @@ def test_UUID_version(self, conversation_id):
     def test_variant(self, conversation_id):
         assert conversation_id[8] >> 6 == 0b10
 
+    def test_correct_timestamp(self, conversation_id):
+        ts = serialization.conversation_id_to_datetime(conversation_id=conversation_id)
+        assert abs(ts - datetime.datetime.now(datetime.timezone.utc)) < datetime.timedelta(hours=1)
+
+
+def test_conversation_id_to_datetime_according_to_uuid_example():
+    """According to the draft https://datatracker.ietf.org/doc/draft-ietf-uuidrev-rfc4122bis/14/
+    017F22E2-79B0-7CC3-98C4-DC0C0C07398F should be a timestamp of
+    Tuesday, February 22, 2022 2:22:22.00 PM GMT-05:00, represented as 1645557742000
+    """
+    cid = uuid.UUID("017F22E2-79B0-7CC3-98C4-DC0C0C07398F").bytes
+    reference = datetime.datetime(
+        2022, 2, 22, 14, 22, 22, tzinfo=datetime.timezone(datetime.timedelta(hours=-5))
+    )
+    ts = serialization.conversation_id_to_datetime(cid)
+    assert reference - ts == datetime.timedelta(0)
+
 
 def test_json_type_result_is_response():
     assert JsonContentTypes.RESPONSE in JsonContentTypes.RESULT_RESPONSE