turret/net_controller.py at master · abyll/turret · GitHub

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
#! /usr/bin/env python
## Networked Turret Controller
# acts as a server

from sys import stdout
from twisted.python.log import startLogging, err

from turret import Turret, TestTurret

from twisted.internet import reactor
from twisted.internet.protocol import Factory
from twisted.internet.endpoints import TCP4ServerEndpoint

from twisted.protocols.amp import Integer, Float, String, Boolean, Command
from twisted.protocols import amp


class TiltCommand(Command):
    arguments = [('speed', Float())]
    requiresAnswer = False

class PanCommand(Command):
    arguments = [('speed', Float())]
    requiresAnswer = False

class StopTiltCommand(Command):
    requiresAnswer = False

class StopPanCommand(Command):
    requiresAnswer = False

class FireCommand(Command):
    requiresAnswer = False

class KeepalivePing(Command):
    pass

class TurretControlProtocol(amp.AMP):
    @TiltCommand.responder
    def tilt(self, speed):
        self.factory.Tilt(speed)
        return {"result": True}

    @PanCommand.responder
    def pan(self, speed):
        self.factory.Pan(speed)
        return {"result": True}

    @StopTiltCommand.responder
    def stop_tilt(self):
        self.factory.StopTilt()
        return {}

    @StopPanCommand.responder
    def stop_pan(self):
        self.factory.StopPan()
        return {}

    @FireCommand.responder
    def fire(self):
        self.factory.Fire()
        return {}

    @KeepalivePing.responder
    def keepalive(self):
        print("alive")
        return {}

    def logout(self):
        self.factory.disconnect()

    def login(self):
        self.factory.login()

    def connectionMade(self):
        self.factory.Calibrate()
        amp.AMP.connectionMade(self)

    def connectionLost(self, reason):
        amp.AMP.connectionLost(self, reason)
        if reason == "logout":
            self.factory.Logout()
        self.factory.StopAll()

class TurretControlFactory(Factory):
    def __init__(self, selected_turret):
        self.connected = False # flag to distinguish intentional and unintentional disconnections
        self.turret = selected_turret()

    def Tilt(self, speed):
        self.turret.tilt(speed)

    def Pan(self, speed):
        self.turret.pan_forever(speed)

    def StopTilt(self):
        self.turret.stop_tilt()

    def StopPan(self):
        self.turret.stop_pan()

    def StopAll(self):
        self.turret.stop_tilt()
        self.turret.stop_pan()

    def Fire(self):
        self.turret.fire()

    def Calibrate(self):
        if not self.connected:
            self.turret.calibrate()

    def disconnect(self):
        self.connected = False

    def login(self):
        self.connected = True

def main():
    startLogging(stdout)
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("-r", "--real", action="store_true") #Use test by default
    args = parser.parse_args()

    if args.real:
        turret = Turret
    else:
        turret = TestTurret

    factory = TurretControlFactory(turret)
    factory.protocol = TurretControlProtocol
    server_endpoint = TCP4ServerEndpoint(reactor, 8750)
    listening_port = server_endpoint.listen(factory)
    print("Running")
    reactor.run()


if __name__ == "__main__":
    main()