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 maybe allow diferent time size (use now for time)
 95    if t.size != TimeSize.SEVEN:
 96        raise ValueError('unsupported time size')
 97
 98    local_dt = datetime.datetime(
 99        year=2000 + t.years if t.years < 70 else 1900 + t.years,
100        month=t.months,
101        day=t.day_of_month,
102        hour=t.hours,
103        minute=t.minutes,
104        second=int(t.milliseconds / 1000),
105        microsecond=(t.milliseconds % 1000) * 1000,
106        fold=not t.summer_time)
107
108    return local_dt.astimezone(tz=datetime.timezone.utc)