hat.drivers.iec60870.encodings.common

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