-
Notifications
You must be signed in to change notification settings - Fork 22
Expand file tree
/
Copy pathsample.py
More file actions
133 lines (107 loc) · 4.02 KB
/
Copy pathsample.py
File metadata and controls
133 lines (107 loc) · 4.02 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
"""
Author: Steffen Vogel <post@steffenvogel.de>
Author: Philipp Jungkamp <Philipp.Jungkamp@opal-rt.com>
SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University
SPDX-License-Identifier: Apache-2.0
""" # noqa: E501
from ctypes import c_double, c_float, sizeof
from dataclasses import dataclass, field
from datetime import datetime, timezone
from functools import total_ordering
from sys import byteorder as native
assert sizeof(c_float) == 4
assert sizeof(c_double) == 8
Signal = bool | int | float | complex
@total_ordering
@dataclass
class Timestamp:
"""
A VILLASnode timestamp. Based on the C struct timespec.
These timestamps are always UTC.
"""
seconds: int
nanoseconds: int = 0
def _as_digest_bytes(self):
sec = self.seconds.to_bytes(8, "little")
nsec = self.nanoseconds.to_bytes(8, "little")
return bytes().join([sec, nsec])
@classmethod
def fromdatetime(cls, ts: datetime) -> "Timestamp":
secs = int(ts.timestamp())
nsecs = int(1000 * ts.microsecond)
return cls(seconds=secs, nanoseconds=nsecs)
@classmethod
def fromtimestamp(cls, ts: float) -> "Timestamp":
secs = int(ts)
nsecs = int(1e9 * (ts - float(secs)))
return cls(seconds=secs, nanoseconds=nsecs)
def timestamp(self) -> float:
return float(self)
def datetime(self) -> datetime:
return datetime.fromtimestamp(self.timestamp(), tz=timezone.utc)
def __float__(self):
return float(self.seconds) + float(self.nanoseconds) * 1e-9
def _as_ordered_tuple(self):
return (
self.seconds,
self.nanoseconds,
)
def __eq__(self, other: object):
if not isinstance(other, Timestamp):
return False
return self._as_ordered_tuple() == other._as_ordered_tuple()
def __lt__(self, other: "Timestamp"):
return self._as_ordered_tuple() < other._as_ordered_tuple()
@total_ordering
@dataclass(kw_only=True)
class Sample:
"""
A VILLASnode sample.
"""
ts_origin: Timestamp | None = None
ts_received: Timestamp | None = None
sequence: int | None = None
new_frame: bool = False
data: list[Signal] = field(default_factory=list)
def _as_ordered_tuple(self):
return (
self.ts_origin is not None,
self.ts_origin if self.ts_origin is not None else Timestamp(0),
self.ts_received is not None,
self.ts_received if self.ts_received is not None else Timestamp(0),
self.sequence is not None,
self.sequence if self.sequence is not None else 0,
not self.new_frame,
self.data,
)
def __eq__(self, other: object):
if not isinstance(other, Sample):
return False
return self._as_ordered_tuple() == other._as_ordered_tuple()
def __lt__(self, other: "Timestamp"):
return self._as_ordered_tuple() < other._as_ordered_tuple()
def _as_digest_bytes(self):
def signal_to_bytes(signal):
match signal:
case bool():
return signal.to_bytes(1, "little")
case int():
return signal.to_bytes(8, "little")
case float():
i = int.from_bytes(bytes(c_double(signal)), native)
return i.to_bytes(8, "little")
case complex():
f_real = signal.real
f_imag = signal.imag
i_real = int.from_bytes(bytes(c_float(f_real)), native)
i_imag = int.from_bytes(bytes(c_float(f_imag)), native)
real = i_real.to_bytes(4, "little")
imag = i_imag.to_bytes(4, "little")
return bytes().join([real, imag])
return bytes().join(
[
self.ts_origin._as_digest_bytes(),
self.sequence.to_bytes(8, "little"),
]
+ list(map(signal_to_bytes, self.data))
)