hat.drivers.iec60870.encodings.encoder

  1import collections
  2import typing
  3
  4from hat import util
  5
  6from hat.drivers.iec60870.encodings import common
  7
  8
  9AsduType: typing.TypeAlias = int
 10AsduTypeTimeSizes: typing.TypeAlias = dict[AsduType, common.TimeSize]
 11DecodeIoElementCb: typing.TypeAlias = typing.Callable[[util.Bytes, AsduType],
 12                                                      tuple[typing.Any,
 13                                                            util.Bytes]]
 14EncodeIoElementCb: typing.TypeAlias = typing.Callable[[typing.Any, AsduType],
 15                                                      typing.Iterable[int]]
 16
 17
 18def decode_time(time_bytes: util.Bytes,
 19                time_size: common.TimeSize
 20                ) -> common.Time:
 21    milliseconds = (time_bytes[1] << 8) | time_bytes[0]
 22    invalid = (bool(time_bytes[2] & 0x80) if time_size.value > 2 else None)
 23    minutes = (time_bytes[2] & 0x3F if time_size.value > 2 else None)
 24    summer_time = (bool(time_bytes[3] & 0x80) if time_size.value > 3 else None)
 25    hours = (time_bytes[3] & 0x1F if time_size.value > 3 else None)
 26    day_of_week = (time_bytes[4] >> 5 if time_size.value > 4 else None)
 27    day_of_month = (time_bytes[4] & 0x1F if time_size.value > 4 else None)
 28    months = (time_bytes[5] & 0x0F if time_size.value > 4 else None)
 29    years = (time_bytes[6] & 0x7F if time_size.value > 4 else None)
 30
 31    return common.Time(size=time_size,
 32                       milliseconds=milliseconds,
 33                       invalid=invalid,
 34                       minutes=minutes,
 35                       summer_time=summer_time,
 36                       hours=hours,
 37                       day_of_week=day_of_week,
 38                       day_of_month=day_of_month,
 39                       months=months,
 40                       years=years)
 41
 42
 43def encode_time(time: common.Time,
 44                time_size: common.TimeSize
 45                ) -> typing.Iterable[int]:
 46    if time_size.value > time.size.value:
 47        raise ValueError('unsupported time size')
 48
 49    yield time.milliseconds & 0xFF
 50    yield (time.milliseconds >> 8) & 0xFF
 51
 52    if time_size.value > 2:
 53        yield ((0x80 if time.invalid else 0) |
 54               (time.minutes & 0x3F))
 55
 56    if time_size.value > 3:
 57        yield ((0x80 if time.summer_time else 0) |
 58               (time.hours & 0x1F))
 59
 60    if time_size.value > 4:
 61        yield (((time.day_of_week & 0x07) << 5) |
 62               (time.day_of_month & 0x1F))
 63        yield time.months & 0x0F
 64        yield time.years & 0x7F
 65
 66
 67class Encoder:
 68
 69    def __init__(self,
 70                 cause_size: common.CauseSize,
 71                 asdu_address_size: common.AsduAddressSize,
 72                 io_address_size: common.IoAddressSize,
 73                 asdu_type_time_sizes: AsduTypeTimeSizes,
 74                 inverted_sequence_bit: bool,
 75                 decode_io_element_cb: DecodeIoElementCb,
 76                 encode_io_element_cb: EncodeIoElementCb):
 77        self._cause_size = cause_size
 78        self._asdu_address_size = asdu_address_size
 79        self._io_address_size = io_address_size
 80        self._asdu_type_time_sizes = asdu_type_time_sizes
 81        self._inverted_sequence_bit = inverted_sequence_bit
 82        self._decode_io_element_cb = decode_io_element_cb
 83        self._encode_io_element_cb = encode_io_element_cb
 84
 85    @property
 86    def cause_size(self) -> common.CauseSize:
 87        return self._cause_size
 88
 89    @property
 90    def asdu_address_size(self) -> common.AsduAddressSize:
 91        return self._asdu_address_size
 92
 93    @property
 94    def io_address_size(self) -> common.IoAddressSize:
 95        return self._io_address_size
 96
 97    def decode_asdu(self,
 98                    asdu_bytes: util.Bytes
 99                    ) -> tuple[common.ASDU, util.Bytes]:
100        asdu_type = asdu_bytes[0]
101        io_number = asdu_bytes[1] & 0x7F
102        is_sequence = bool(asdu_bytes[1] & 0x80)
103        if self._inverted_sequence_bit:
104            is_sequence = not is_sequence
105        io_count = 1 if is_sequence else io_number
106        ioe_element_count = io_number if is_sequence else 1
107
108        rest = asdu_bytes[2:]
109        cause, rest = _decode_int(rest, self._cause_size.value)
110        address, rest = _decode_int(rest, self._asdu_address_size.value)
111
112        ios = collections.deque()
113        for _ in range(io_count):
114            io, rest = self._decode_io(asdu_type, ioe_element_count, rest)
115            ios.append(io)
116
117        asdu = common.ASDU(type=asdu_type,
118                           cause=cause,
119                           address=address,
120                           ios=list(ios))
121        return asdu, rest
122
123    def encode_asdu(self, asdu: common.ASDU) -> util.Bytes:
124        data = collections.deque()
125        data.append(asdu.type)
126
127        is_sequence = len(asdu.ios) == 1 and len(asdu.ios[0].elements) > 1
128        if is_sequence:
129            data.append(
130                (0x00 if self._inverted_sequence_bit else 0x80) |
131                len(asdu.ios[0].elements))
132        else:
133            data.append(
134                (0x80 if self._inverted_sequence_bit else 0x00) |
135                len(asdu.ios))
136
137        data.extend(_encode_int(asdu.cause, self._cause_size.value))
138        data.extend(_encode_int(asdu.address, self._asdu_address_size.value))
139
140        for io in asdu.ios:
141            if not is_sequence and len(io.elements) != 1:
142                raise ValueError('invalid number of IO elements')
143            data.extend(self._encode_io(asdu.type, io))
144
145        return bytes(data)
146
147    def _decode_io(self, asdu_type, ioe_element_count, io_bytes):
148        address, rest = _decode_int(io_bytes, self._io_address_size.value)
149
150        elements = collections.deque()
151        for _ in range(ioe_element_count):
152            element, rest = self._decode_io_element_cb(rest, asdu_type)
153            elements.append(element)
154
155        time_size = self._asdu_type_time_sizes.get(asdu_type)
156        if time_size:
157            time, rest = decode_time(rest, time_size), rest[time_size.value:]
158        else:
159            time = None
160
161        io = common.IO(address=address,
162                       elements=list(elements),
163                       time=time)
164        return io, rest
165
166    def _encode_io(self, asdu_type, io):
167        yield from _encode_int(io.address, self._io_address_size.value)
168
169        for element in io.elements:
170            yield from self._encode_io_element_cb(element, asdu_type)
171
172        time_size = self._asdu_type_time_sizes.get(asdu_type)
173        if time_size:
174            yield from encode_time(io.time, time_size)
175
176
177def _decode_int(data, size):
178    return int.from_bytes(data[:size], 'little'), data[size:]
179
180
181def _encode_int(x, size):
182    return x.to_bytes(size, 'little')