2024-04-26 10:01:29 +02:00
|
|
|
#!/usr/bin/env python3
|
2024-05-25 13:46:14 +02:00
|
|
|
import enum
|
2024-04-26 10:01:29 +02:00
|
|
|
from socket import AF_INET, SOCK_DGRAM, socket
|
2024-05-25 13:46:14 +02:00
|
|
|
from pydantic import BaseModel
|
2024-04-26 10:01:29 +02:00
|
|
|
import msgpack
|
|
|
|
|
2024-05-25 13:08:32 +02:00
|
|
|
|
|
|
|
TEST_PERSON = {
|
2024-04-26 10:01:29 +02:00
|
|
|
"age": 24,
|
|
|
|
"name": "Nadine",
|
|
|
|
}
|
|
|
|
|
2024-05-25 13:08:32 +02:00
|
|
|
|
2024-05-25 13:46:14 +02:00
|
|
|
class Devices(str, enum.Enum):
|
|
|
|
MGT = "Mgt"
|
|
|
|
MGM = "Mgm"
|
|
|
|
|
|
|
|
|
|
|
|
class SwitchState(str, enum.Enum):
|
|
|
|
OFF = "Off"
|
|
|
|
ON = "On"
|
|
|
|
UNKNOWN = "Unknown"
|
|
|
|
FAULTY = "Faulty"
|
|
|
|
|
|
|
|
|
|
|
|
class SwitchMap(BaseModel):
|
|
|
|
valid: bool
|
|
|
|
switch_map: dict[Devices, SwitchState]
|
|
|
|
|
|
|
|
|
2024-05-25 13:08:32 +02:00
|
|
|
def msg_pack_unloading(recv_back: bytes):
|
|
|
|
unpacked = msgpack.unpackb(recv_back)
|
|
|
|
print(f"unpacked: {unpacked}")
|
|
|
|
loaded_back = msgpack.loads(recv_back)
|
|
|
|
print(loaded_back)
|
|
|
|
|
|
|
|
|
|
|
|
def main():
|
|
|
|
server_socket = socket(AF_INET, SOCK_DGRAM)
|
|
|
|
target_address = "localhost", 7301
|
|
|
|
msg_pack_stuff = msgpack.packb(TEST_PERSON)
|
|
|
|
assert msg_pack_stuff is not None
|
|
|
|
_ = server_socket.sendto(msg_pack_stuff, target_address)
|
|
|
|
recv_back = server_socket.recv(4096)
|
|
|
|
print(f"recv back: {recv_back}")
|
2024-05-25 13:46:14 +02:00
|
|
|
switch_map = SwitchMap.model_validate_json(recv_back)
|
|
|
|
print(f"switch map: {switch_map}")
|
2024-05-25 13:08:32 +02:00
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
main()
|