hat.drivers.iec60870.encodings.common

  1import datetime
  2import enum
  3import time
  4import typing
  5
  6
  7class CauseSize(enum.Enum):
  8    ONE = 1
  9    TWO = 2
 10
 11
 12class AsduAddressSize(enum.Enum):
 13    ONE = 1
 14    TWO = 2
 15
 16
 17class IoAddressSize(enum.Enum):
 18    ONE = 1
 19    TWO = 2
 20    THREE = 3
 21
 22
 23class TimeSize(enum.Enum):
 24    TWO = 2
 25    THREE = 3
 26    FOUR = 4
 27    SEVEN = 7
 28
 29
 30class Time(typing.NamedTuple):
 31    size: TimeSize
 32    milliseconds: int
 33    """milliseconds in range [0, 59999]"""
 34    invalid: bool | None
 35    """available for size THREE, FOUR, SEVEN"""
 36    minutes: int | None
 37    """available for size THREE, FOUR, SEVEN (minutes in range [0, 59])"""
 38    summer_time: bool | None
 39    """available for size FOUR, SEVEN"""
 40    hours: int | None
 41    """available for size FOUR, SEVEN (hours in range [0, 23])"""
 42    day_of_week: int | None
 43    """available for size SEVEN (day_of_week in range [1, 7])"""
 44    day_of_month: int | None
 45    """available for size SEVEN (day_of_month in range [1, 31])"""
 46    months: int | None
 47    """available for size SEVEN (months in range [1, 12])"""
 48    years: int | None
 49    """available for size SEVEN (years in range [0, 99])"""
 50
 51
 52class IO(typing.NamedTuple):
 53    address: int
 54    elements: list
 55    time: Time | None
 56
 57
 58class ASDU(typing.NamedTuple):
 59    type: int
 60    cause: int
 61    address: int
 62    ios: list[IO]
 63
 64
 65def time_from_datetime(dt: datetime.datetime,
 66                       invalid: bool = False
 67                       ) -> Time:
 68    """Create Time from datetime.datetime"""
 69    # TODO document edge cases (local time, os implementation, ...)
 70    #  rounding microseconds to the nearest millisecond
 71    dt_rounded = (
 72        dt.replace(microsecond=0) +
 73        datetime.timedelta(milliseconds=round(dt.microsecond / 1000)))
 74    local_time = time.localtime(dt_rounded.timestamp())
 75
 76    return Time(
 77        size=TimeSize.SEVEN,
 78        milliseconds=(local_time.tm_sec * 1000 +
 79                      dt_rounded.microsecond // 1000),
 80        invalid=invalid,
 81        minutes=local_time.tm_min,
 82        summer_time=bool(local_time.tm_isdst),
 83        hours=local_time.tm_hour,
 84        day_of_week=local_time.tm_wday + 1,
 85        day_of_month=local_time.tm_mday,
 86        months=local_time.tm_mon,
 87        years=local_time.tm_year % 100)
 88
 89
 90def time_to_datetime(t: Time
 91                     ) -> datetime.datetime:
 92    """Convert Time to datetime.datetime"""
 93    # TODO document edge cases (local time, os implementation, ...)
 94    # TODO support TimeSize.FOUR
 95    if t.size == TimeSize.TWO:
 96        local_now = datetime.datetime.now()
 97        local_dt = local_now.replace(
 98            second=int(t.milliseconds / 1000),
 99            microsecond=(t.milliseconds % 1000) * 1000)
100
101        local_seconds = local_now.second + local_now.microsecond / 1_000_000
102        t_seconds = t.milliseconds / 1_000
103
104        if abs(local_seconds - t_seconds) > 30:
105            if local_seconds < t_seconds:
106                local_dt = local_dt - datetime.timedelta(minutes=1)
107
108            else:
109                local_dt = local_dt + datetime.timedelta(minutes=1)
110
111    elif t.size == TimeSize.THREE:
112        local_now = datetime.datetime.now()
113        local_dt = local_now.replace(
114            minute=t.minutes,
115            second=int(t.milliseconds / 1000),
116            microsecond=(t.milliseconds % 1000) * 1000)
117
118        local_minutes = (local_now.minute +
119                         local_now.second / 60 +
120                         local_now.microsecond / 60_000_000)
121        t_minutes = t.minutes + t.milliseconds / 60_000
122
123        if abs(local_minutes - t_minutes) > 30:
124            if local_minutes < t_minutes:
125                local_dt = local_dt - datetime.timedelta(hours=1)
126
127            else:
128                local_dt = local_dt + datetime.timedelta(hours=1)
129
130    elif t.size == TimeSize.SEVEN:
131        local_dt = datetime.datetime(
132            year=2000 + t.years if t.years < 70 else 1900 + t.years,
133            month=t.months,
134            day=t.day_of_month,
135            hour=t.hours,
136            minute=t.minutes,
137            second=int(t.milliseconds / 1000),
138            microsecond=(t.milliseconds % 1000) * 1000,
139            fold=not t.summer_time)
140
141    else:
142        raise ValueError('unsupported time size')
143
144    return local_dt.astimezone(tz=datetime.timezone.utc)