|
| 1 | +import random |
| 2 | +import struct |
| 3 | +from typing import Generic, Literal, Optional, Type, TypeVar |
| 4 | + |
| 5 | +from PyCRC.CRCCCITT import CRCCCITT as CRC |
| 6 | + |
| 7 | +PARAM_CMDS = ["VS", "?VR"] |
| 8 | +FloatOrInt = TypeVar("FloatOrInt", float, int) |
| 9 | +ParamCmds = Literal["VS", "?VR"] |
| 10 | + |
| 11 | + |
| 12 | +def calc_checksum(string: str) -> str: |
| 13 | + """Calculate CRC checksum.""" |
| 14 | + return f"{CRC().calculate(string):04X}" |
| 15 | + |
| 16 | + |
| 17 | +def construct_param_cmd( |
| 18 | + device_addr: int, |
| 19 | + cmd: str, |
| 20 | + param_id: int, |
| 21 | + value_type: Type[FloatOrInt], |
| 22 | + param_inst: int = 1, |
| 23 | + value: Optional[FloatOrInt] = None, |
| 24 | + seq_num: Optional[int] = None, |
| 25 | +) -> str: |
| 26 | + """ |
| 27 | + Construct a MeCom command. |
| 28 | +
|
| 29 | + :param device_addr: Device address (0 .. 255). Broadcast Device Address (0) will |
| 30 | + send the command to all connected Meerstetter devices |
| 31 | + :param param_id: Parameter ID (0 .. 65535) |
| 32 | + :param value_type: Value type (int or float) |
| 33 | + :param param_inst: Parameter instance (0 .. 255). For most parameters the instance |
| 34 | + is used to address the channel on the device |
| 35 | + :param value: Value to set |
| 36 | + :param seq_num: Sequence number (0 .. 65535). If not given, a random number will be |
| 37 | + generated |
| 38 | + :return: MeCom command |
| 39 | + """ |
| 40 | + if seq_num is None: |
| 41 | + seq_num = random.randint(0, 65535) |
| 42 | + |
| 43 | + if seq_num < 0 or seq_num > 65535: |
| 44 | + raise ValueError("seq_num must be between 0 and 65535") |
| 45 | + |
| 46 | + if cmd not in PARAM_CMDS: |
| 47 | + raise ValueError(f"cmd must be one of {PARAM_CMDS}") |
| 48 | + |
| 49 | + if device_addr < 0 or device_addr > 255: |
| 50 | + raise ValueError("device_addr must be between 0 and 255") |
| 51 | + |
| 52 | + if cmd in ["VS", "?VR"] and param_id is None: |
| 53 | + raise ValueError("param_id must be given for VS and ?VR commands") |
| 54 | + |
| 55 | + if cmd == "VS": |
| 56 | + if value is None: |
| 57 | + raise ValueError("value must be given for VS command") |
| 58 | + if value_type is float: |
| 59 | + # convert float to hex of length 8, remove the leading '0X' and capitalize |
| 60 | + val = hex(struct.unpack("<I", struct.pack("<f", value))[0])[2:].upper() |
| 61 | + elif value_type is int: |
| 62 | + # convert int to hex of length 8 |
| 63 | + val = f"{value:08X}" |
| 64 | + elif cmd == "?VR": |
| 65 | + val = "" |
| 66 | + |
| 67 | + cmd = f"#{device_addr:02X}{seq_num:04X}{cmd}{param_id:04X}{param_inst:02X}{val}" |
| 68 | + return f"{cmd}{calc_checksum(cmd)}\r" |
| 69 | + |
| 70 | + |
| 71 | +def construct_reset_cmd(device_addr: int, seq_num: Optional[int] = None) -> str: |
| 72 | + """ |
| 73 | + Construct a MeCom reset command. |
| 74 | +
|
| 75 | + :param device_addr: Device address (0 .. 255). Broadcast Device Address (0) will |
| 76 | + send the command to all connected Meerstetter devices |
| 77 | + :param seq_num: Sequence number (0 .. 65535). If not given, a random number will be |
| 78 | + generated |
| 79 | + :return: MeCom command |
| 80 | + """ |
| 81 | + if seq_num is None: |
| 82 | + seq_num = random.randint(0, 65535) |
| 83 | + |
| 84 | + if seq_num < 0 or seq_num > 65535: |
| 85 | + raise ValueError("seq_num must be between 0 and 65535") |
| 86 | + |
| 87 | + cmd = f"#{device_addr:02X}{seq_num:04X}RS" |
| 88 | + return f"{cmd}{calc_checksum(cmd)}\r" |
| 89 | + |
| 90 | + |
| 91 | +def verify_response(reponse: "Message", request: "Message") -> bool: |
| 92 | + """ |
| 93 | + Verify a MeCom response. |
| 94 | +
|
| 95 | + :param reponse: MeCom response |
| 96 | + :param request: MeCom request |
| 97 | + :return: True if response is valid, False otherwise |
| 98 | + """ |
| 99 | + checksum_correct = reponse.checksum == calc_checksum(reponse[0:-5]) |
| 100 | + request_match = reponse.seq_num == request.seq_num |
| 101 | + return checksum_correct & request_match |
| 102 | + |
| 103 | + |
| 104 | +class Message(str, Generic[FloatOrInt]): |
| 105 | + value_type: Type[FloatOrInt] |
| 106 | + |
| 107 | + def __new__(cls, response: str, value_type: Type[FloatOrInt]): |
| 108 | + return super().__new__(cls, response) |
| 109 | + |
| 110 | + def __init__(self, response: str, value_type: Type[FloatOrInt]) -> None: |
| 111 | + self.value_type = value_type |
| 112 | + self.device_addr = int(self[1:3], 8) |
| 113 | + self.seq_num = int(self[3:7], 16) |
| 114 | + self.payload = self[7:-5] |
| 115 | + self.checksum = self[-5:-1] |
| 116 | + |
| 117 | + @property |
| 118 | + def value(self) -> FloatOrInt: |
| 119 | + if self.value_type is int: |
| 120 | + return int(self.payload, 16) |
| 121 | + if self.value_type is float: |
| 122 | + return struct.unpack("!f", bytes.fromhex(self.payload))[0] |
| 123 | + else: |
| 124 | + raise ValueError("value_type must be int or float") |
0 commit comments