-
Notifications
You must be signed in to change notification settings - Fork 2
/
sp_id.py
39 lines (30 loc) · 1.24 KB
/
sp_id.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
"""Sparkplug ID management."""
from typing import Optional
from . import sp_helpers
class SPId:
"""Handles the ID of a Sparkplug EoN/device."""
def __init__(self, group_id: str, eon_id: str, dev_id: Optional[str] = None):
self.group_id = group_id
self.eon_id = eon_id
self.dev_id = dev_id
@staticmethod
def from_str(topic: str) -> Optional[str]:
"""Create a SPId object from a topic string"""
tokens = topic.split("/")
if (len(tokens) == 4 or len(tokens) == 5) and tokens[0] == sp_helpers.SP_NAMESPACE:
del tokens[2]
return SPId(*tokens[1:])
return None
def __str__(self) -> str:
"""Returns a string representation of the Id."""
return f"{self.group_id}/{self.eon_id}{sp_helpers.get_dev_id_str(self.dev_id)}"
def __eq__(self, other) -> bool:
return ((self.group_id == other.group_id)
and (self.eon_id == other.eon_id)
and (self.dev_id == other.dev_id))
def is_eon(self) -> bool:
"""Returns True if this Id belongs to an EoN."""
return self.dev_id is None
def is_dev(self) -> bool:
"""Returns True if this Id belongs to a Device"""
return not self.is_eon()