dani/django-ram
1
0
Fork 0
mirror of https://github.com/daniviga/django-ram.git synced 2025-08-05 05:37:50 +02:00
Code Issues Packages Projects Releases Wiki Activity

Run black

Browse Source
This commit is contained in:
Daniele Viganò
2022-04-02 22:58:42 +02:00
parent 6478800f77
commit b598e38f70
27 changed files with 318 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
dcc/driver/apps.py
View File

@@ -3,9 +3,10 @@ from health_check.plugins import plugin_dir
class DriverConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'driver'
default_auto_field = "django.db.models.BigAutoField"
name = "driver"
def ready(self):
from driver.health import DriverHealthCheck
plugin_dir.register(DriverHealthCheck)

14
dcc/driver/connector.py
View File

@@ -8,7 +8,7 @@ class Connector:
self.config = DriverConfiguration.get_solo()
def __send_data(self, message):
resp = b''
resp = b""
# convert to binary if str is received
if isinstance(message, str):
message = message.encode()
@@ -30,10 +30,12 @@ class Connector:
def ops(self, address, data, function=False):
if function:
message = "<F {0} {1} {2}>".format(
address, data['function'], data['state'])
address, data["function"], data["state"]
)
else:
message = "<t 1 {0} {1} {2}>".format(
address, data['speed'], data['direction'])
address, data["speed"], data["direction"]
)
self.__send_data(message)
def infra(self, data):
@@ -43,9 +45,9 @@ class Connector:
track = ""
if data["power"]:
self.__send_data('<1{}>'.format(track))
self.__send_data("<1{}>".format(track))
else:
self.__send_data('<0{}>'.format(track))
self.__send_data("<0{}>".format(track))
def emergency(self):
self.__send_data('<!>')
self.__send_data("<!>")

8
dcc/driver/health.py
View File

@@ -1,6 +1,8 @@
from health_check.backends import BaseHealthCheckBackend
from health_check.exceptions import (ServiceUnavailable,
ServiceReturnedUnexpectedResult)
from health_check.exceptions import (
ServiceUnavailable,
ServiceReturnedUnexpectedResult,
)
from driver.connector import Connector
@@ -10,7 +12,7 @@ class DriverHealthCheck(BaseHealthCheckBackend):
def check_status(self):
try:
Connector().passthrough(b'<s>')
Connector().passthrough(b"<s>")
except ConnectionRefusedError as e:
self.add_error(ServiceUnavailable("IOError"), e)
except Exception as e:

3
dcc/driver/models.py
View File

@@ -4,7 +4,8 @@ from solo.models import SingletonModel
class DriverConfiguration(SingletonModel):
remote_host = models.GenericIPAddressField(
protocol="IPv4", default="192.168.4.1")
protocol="IPv4", default="192.168.4.1"
)
remote_port = models.SmallIntegerField(default=2560)
timeout = models.SmallIntegerField(default=250)

5
dcc/driver/serializers.py
View File

@@ -14,5 +14,6 @@ class CabSerializer(serializers.Serializer):
class InfraSerializer(serializers.Serializer):
power = serializers.BooleanField(required=True)
track = serializers.ChoiceField(
choices=('main', 'prog', 'join', 'MAIN', 'PROG', 'JOIN'),
required=False)
choices=("main", "prog", "join", "MAIN", "PROG", "JOIN"),
required=False,
)

12
dcc/driver/urls.py
View File

@@ -2,10 +2,10 @@ from django.urls import path
from driver.views import SendCommand, Function, Cab, Emergency, Infra, Test
urlpatterns = [
path('test', Test.as_view()),
path('emergency', Emergency.as_view()),
path('infra', Infra.as_view()),
path('command', SendCommand.as_view()),
path('<int:address>/cab', Cab.as_view()),
path('<int:address>/function', Function.as_view()),
path("test", Test.as_view()),
path("emergency", Emergency.as_view()),
path("infra", Infra.as_view()),
path("command", SendCommand.as_view()),
path("<int:address>/cab", Cab.as_view()),
path("<int:address>/function", Function.as_view()),
]

61
dcc/driver/views.py
View File

@@ -7,7 +7,10 @@ from rest_framework.response import Response
from dcc.parsers import PlainTextParser
from driver.connector import Connector
from driver.serializers import (
FunctionSerializer, CabSerializer, InfraSerializer)
FunctionSerializer,
CabSerializer,
InfraSerializer,
)
from roster.models import RollingStock
@@ -15,10 +18,12 @@ def addresschecker(f):
"""
Check if DCC address does exist in the database
"""
def addresslookup(request, address, *args):
if not RollingStock.objects.filter(address=address):
raise Http404
return f(request, address, *args)
return addresslookup
@@ -26,32 +31,38 @@ class Test(APIView):
"""
Send a test <s> command
"""
parser_classes = [PlainTextParser]
def get(self, request):
response = Connector().passthrough("<s>")
return Response({"response": response.decode()},
status=status.HTTP_202_ACCEPTED)
return Response(
{"response": response.decode()}, status=status.HTTP_202_ACCEPTED
)
class SendCommand(APIView):
"""
Command passthrough
"""
parser_classes = [PlainTextParser]
def put(self, request):
data = request.data
if not data:
raise serializers.ValidationError({
"error": "a string is expected"})
raise serializers.ValidationError(
{"error": "a string is expected"}
)
cmd = data.decode().strip()
if not (cmd.startswith("<") and cmd.endswith(">")):
raise serializers.ValidationError({
"error": "please provide a valid command"})
raise serializers.ValidationError(
{"error": "please provide a valid command"}
)
response = Connector().passthrough(cmd)
return Response({"response": response.decode()},
status=status.HTTP_202_ACCEPTED)
return Response(
{"response": response.decode()}, status=status.HTTP_202_ACCEPTED
)
@method_decorator(addresschecker, name="put")
@@ -59,15 +70,14 @@ class Function(APIView):
"""
Send "Function" commands to a valid DCC address
"""
def put(self, request, address):
serializer = FunctionSerializer(data=request.data)
if serializer.is_valid():
Connector().ops(address, serializer.data, function=True)
return Response(serializer.data,
status=status.HTTP_202_ACCEPTED)
return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
return Response(serializer.errors,
status=status.HTTP_400_BAD_REQUEST)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
@method_decorator(addresschecker, name="put")
@@ -75,42 +85,43 @@ class Cab(APIView):
"""
Send "Cab" commands to a valid DCC address
"""
def put(self, request, address):
serializer = CabSerializer(data=request.data)
if serializer.is_valid():
Connector().ops(address, serializer.data)
return Response(serializer.data,
status=status.HTTP_202_ACCEPTED)
return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
return Response(serializer.errors,
status=status.HTTP_400_BAD_REQUEST)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
class Infra(APIView):
"""
Send "Infra" commands to a valid DCC address
"""
def put(self, request):
serializer = InfraSerializer(data=request.data)
if serializer.is_valid():
Connector().infra(serializer.data)
return Response(serializer.data,
status=status.HTTP_202_ACCEPTED)
return Response(serializer.data, status=status.HTTP_202_ACCEPTED)
return Response(serializer.errors,
status=status.HTTP_400_BAD_REQUEST)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
class Emergency(APIView):
"""
Send an "Emergency" stop, no matter the HTTP method used
"""
def put(self, request):
Connector().emergency()
return Response({"response": "emergency stop"},
status=status.HTTP_202_ACCEPTED)
return Response(
{"response": "emergency stop"}, status=status.HTTP_202_ACCEPTED
)
def get(self, request):
Connector().emergency()
return Response({"response": "emergency stop"},
status=status.HTTP_202_ACCEPTED)
return Response(
{"response": "emergency stop"}, status=status.HTTP_202_ACCEPTED
)