2025-summer/team-3
12
Fork 0
Code Issues Pull requests Projects Releases Activity

rename Message → MeshMessage

Browse source
This commit is contained in:
Laura Klünder 2023-10-04 15:44:54 +02:00
parent 21b75bec86
commit c5ac3ceb76
6 changed files with 34 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/c3nav/control/forms.py
View file

@ -20,7 +20,7 @@ from c3nav.mapdata.forms import I18nModelFormMixin
from c3nav.mapdata.models import MapUpdate, Space
from c3nav.mapdata.models.access import (AccessPermission, AccessPermissionToken, AccessPermissionTokenItem,
AccessRestriction, AccessRestrictionGroup)
from c3nav.mesh.messages import MessageType
from c3nav.mesh.messages import MeshMessageType
from c3nav.mesh.models import MeshNode
from c3nav.site.models import Announcement
@ -302,7 +302,7 @@ class MapUpdateForm(ModelForm):
class MeshMessageFilterForm(Form):
message_types = MultipleChoiceField(
choices=[(msgtype.value, msgtype.name) for msgtype in MessageType],
choices=[(msgtype.value, msgtype.name) for msgtype in MeshMessageType],
required=False,
label=_('message types'),
)

4
src/c3nav/control/views/mesh.py
View file

@ -8,7 +8,7 @@ from django.views.generic import ListView, DetailView, FormView
from c3nav.control.forms import MeshMessageFilterForm
from c3nav.control.views.base import ControlPanelMixin
from c3nav.mesh.forms import MeshMessageForm
from c3nav.mesh.messages import MessageType
from c3nav.mesh.messages import MeshMessageType
from c3nav.mesh.models import MeshNode, NodeMessage
@ -69,7 +69,7 @@ class MeshMessageSendView(ControlPanelMixin, FormView):
def get_form_class(self):
try:
return MeshMessageForm.get_form_for_type(MessageType[self.kwargs['msg_type']])
return MeshMessageForm.get_form_for_type(MeshMessageType[self.kwargs['msg_type']])
except KeyError:
raise Http404('unknown message type')

8
src/c3nav/mesh/consumers.py
View file

@ -6,7 +6,7 @@ from django.utils import timezone
from c3nav.control.views.utils import get_mesh_comm_group
from c3nav.mesh import messages
from c3nav.mesh.messages import Message, BROADCAST_ADDRESS
from c3nav.mesh.messages import MeshMessage, BROADCAST_ADDRESS
from c3nav.mesh.models import MeshNode, NodeMessage
@ -36,7 +36,7 @@ class MeshConsumer(WebsocketConsumer):
if bytes_data is None:
return
try:
msg = messages.Message.decode(bytes_data)
msg = messages.MeshMessage.decode(bytes_data)
except Exception:
traceback.print_exc()
return
@ -96,9 +96,9 @@ class MeshConsumer(WebsocketConsumer):
self.remove_dst_nodes((data["address"], ))
def mesh_send(self, data):
self.send_msg(Message.fromjson(data["msg"]))
self.send_msg(MeshMessage.fromjson(data["msg"]))
def log_received_message(self, src_node: MeshNode, msg: messages.Message):
def log_received_message(self, src_node: MeshNode, msg: messages.MeshMessage):
NodeMessage.objects.create(
uplink_node=self.uplink_node,
src_node=src_node,

6
src/c3nav/mesh/forms.py
View file

@ -2,7 +2,7 @@ from django import forms
from django.http import Http404
from django.utils.translation import gettext_lazy as _
from c3nav.mesh.messages import MessageType, Message, ROOT_ADDRESS
from c3nav.mesh.messages import MeshMessageType, MeshMessage, ROOT_ADDRESS
from c3nav.mesh.models import MeshNode
@ -56,7 +56,7 @@ class MeshMessageForm(forms.Form):
class ConfigUplinkMessageForm(MeshMessageForm):
msg_type = MessageType.CONFIG_UPLINK
msg_type = MeshMessageType.CONFIG_UPLINK
enabled = forms.BooleanField(required=False, label=_('enabled'))
ssid = forms.CharField(required=False, label=_('ssid'), max_length=31)
@ -80,7 +80,7 @@ class ConfigUplinkMessageForm(MeshMessageForm):
recipients = [self.recipient] if self.recipient else self.cleaned_data['recipients']
for recipient in recipients:
print('sending to ', recipient)
Message.fromjson({
MeshMessage.fromjson({
'dst': recipient,
**msg_data,
}).send()

30
src/c3nav/mesh/messages.py
View file

@ -16,7 +16,7 @@ NO_LAYER = 0xFF
@unique
class MessageType(IntEnum):
class MeshMessageType(IntEnum):
ECHO_REQUEST = 0x01
ECHO_RESPONSE = 0x02
@ -42,7 +42,7 @@ class ChipType(IntEnum):
@dataclass
class Message:
class MeshMessage:
dst: str = field(metadata={'format': MacAddressFormat()})
src: str = field(metadata={'format': MacAddressFormat()})
msg_id: int = field(metadata={'format': SimpleFormat('B')}, init=False, repr=False)
@ -53,9 +53,9 @@ class Message:
super().__init_subclass__(**kwargs)
if msg_id:
cls.msg_id = msg_id
if msg_id in Message.msg_types:
if msg_id in MeshMessage.msg_types:
raise TypeError('duplicate use of msg_id %d' % msg_id)
Message.msg_types[msg_id] = cls
MeshMessage.msg_types[msg_id] = cls
def encode(self):
data = bytes()
@ -92,42 +92,42 @@ class Message:
@dataclass
class EchoRequestMessage(Message, msg_id=MessageType.ECHO_REQUEST):
class EchoRequestMessage(MeshMessage, msg_id=MeshMessageType.ECHO_REQUEST):
content: str = field(default='', metadata={'format': VarStrFormat()})
@dataclass
class EchoResponseMessage(Message, msg_id=MessageType.ECHO_RESPONSE):
class EchoResponseMessage(MeshMessage, msg_id=MeshMessageType.ECHO_RESPONSE):
content: str = field(default='', metadata={'format': VarStrFormat()})
@dataclass
class MeshSigninMessage(Message, msg_id=MessageType.MESH_SIGNIN):
class MeshSigninMessage(MeshMessage, msg_id=MeshMessageType.MESH_SIGNIN):
pass
@dataclass
class MeshLayerAnnounceMessage(Message, msg_id=MessageType.MESH_LAYER_ANNOUNCE):
class MeshLayerAnnounceMessage(MeshMessage, msg_id=MeshMessageType.MESH_LAYER_ANNOUNCE):
layer: int = field(metadata={'format': SimpleFormat('B')})
@dataclass
class MeshAddDestinationsMessage(Message, msg_id=MessageType.MESH_ADD_DESTINATIONS):
class MeshAddDestinationsMessage(MeshMessage, msg_id=MeshMessageType.MESH_ADD_DESTINATIONS):
mac_addresses: list[str] = field(default_factory=list, metadata={'format': MacAddressesListFormat()})
@dataclass
class MeshRemoveDestinationsMessage(Message, msg_id=MessageType.MESH_REMOVE_DESTINATIONS):
class MeshRemoveDestinationsMessage(MeshMessage, msg_id=MeshMessageType.MESH_REMOVE_DESTINATIONS):
mac_addresses: list[str] = field(default_factory=list, metadata={'format': MacAddressesListFormat()})
@dataclass
class ConfigDumpMessage(Message, msg_id=MessageType.CONFIG_DUMP):
class ConfigDumpMessage(MeshMessage, msg_id=MeshMessageType.CONFIG_DUMP):
pass
@dataclass
class ConfigFirmwareMessage(Message, msg_id=MessageType.CONFIG_FIRMWARE):
class ConfigFirmwareMessage(MeshMessage, msg_id=MeshMessageType.CONFIG_FIRMWARE):
chip: int = field(metadata={'format': SimpleFormat('H')})
revision: int = field(metadata={'format': SimpleFormat('2B')})
magic_word: int = field(metadata={'format': SimpleFormat('I')}, repr=False)
@ -155,19 +155,19 @@ class ConfigFirmwareMessage(Message, msg_id=MessageType.CONFIG_FIRMWARE):
@dataclass
class ConfigPositionMessage(Message, msg_id=MessageType.CONFIG_POSITION):
class ConfigPositionMessage(MeshMessage, msg_id=MeshMessageType.CONFIG_POSITION):
x_pos: int = field(metadata={'format': SimpleFormat('I')})
y_pos: int = field(metadata={'format': SimpleFormat('I')})
z_pos: int = field(metadata={'format': SimpleFormat('H')})
@dataclass
class ConfigLedMessage(Message, msg_id=MessageType.CONFIG_LED):
class ConfigLedMessage(MeshMessage, msg_id=MeshMessageType.CONFIG_LED):
led_config: LedConfig = field(metadata={'format': LedConfigFormat()})
@dataclass
class ConfigUplinkMessage(Message, msg_id=MessageType.CONFIG_UPLINK):
class ConfigUplinkMessage(MeshMessage, msg_id=MeshMessageType.CONFIG_UPLINK):
enabled: bool = field(metadata={'format': BoolFormat()})
ssid: str = field(metadata={'format': FixedStrFormat(32)})
password: str = field(metadata={'format': FixedStrFormat(64)})

16
src/c3nav/mesh/models.py
View file

@ -4,7 +4,7 @@ from functools import cached_property
from django.db import models, NotSupportedError
from django.utils.translation import gettext_lazy as _
from c3nav.mesh.messages import MessageType, ChipType, Message as MeshMessage
from c3nav.mesh.messages import MeshMessageType, ChipType, MeshMessage as MeshMessage
class MeshNodeQuerySet(models.QuerySet):
@ -18,10 +18,10 @@ class MeshNodeQuerySet(models.QuerySet):
clone._prefetch_last_messages = self._prefetch_last_messages
return clone
def prefetch_last_messages(self, *types: MessageType):
def prefetch_last_messages(self, *types: MeshMessageType):
clone = self._chain()
clone._prefetch_last_messages |= (
set(types) if types else set(msgtype.value for msgtype in MessageType)
set(types) if types else set(msgtype.value for msgtype in MeshMessageType)
)
return clone
@ -37,7 +37,7 @@ class MeshNodeQuerySet(models.QuerySet):
nodes[message.node].last_messages[message.message_type] = message
except NotSupportedError:
pass
print(tuple(nodes.values())[0].last_messages[MessageType.MESH_SIGNIN])
print(tuple(nodes.values())[0].last_messages[MeshMessageType.MESH_SIGNIN])
class LastMessagesByTypeLookup(UserDict):
@ -46,14 +46,14 @@ class LastMessagesByTypeLookup(UserDict):
self.node = node
def _get_key(self, item):
if isinstance(item, MessageType):
if isinstance(item, MeshMessageType):
return item
if isinstance(item, str):
try:
return getattr(MessageType, item)
return getattr(MeshMessageType, item)
except AttributeError:
pass
return MessageType(item)
return MeshMessageType(item)
def __getitem__(self, key):
key = self._get_key(key)
@ -89,7 +89,7 @@ class MeshNode(models.Model):
class NodeMessage(models.Model):
MESSAGE_TYPES = [(msgtype.value, msgtype.name) for msgtype in MessageType]
MESSAGE_TYPES = [(msgtype.value, msgtype.name) for msgtype in MeshMessageType]
src_node = models.ForeignKey('MeshNode', models.PROTECT,
related_name='received_messages', verbose_name=_('node'))
uplink_node = models.ForeignKey('MeshNode', models.PROTECT,