pyduino.slave

  1#Meant to be used in the raspberries
  2from crypt import methods
  3from urllib import response
  4from flask import Flask, request, jsonify
  5from serial import Serial
  6import serial.tools.list_ports as list_ports
  7from time import sleep
  8from pyduino.utils import bcolors
  9import logging
 10import re
 11import os
 12import socket
 13from typing import Any, Optional, Dict
 14
 15TIMEOUT = 60
 16HEADER_DELAY = 5
 17
 18class ReactorServer(Flask):
 19    """
 20    Slave of HTTP Server to Serial handler.
 21    """
 22
 23    def __init__(self, serial_port: Optional[str] = None, baudrate: int = 9600, *args: Any, **kwargs: Any):
 24        """
 25        Initializes the ReactorServer.
 26
 27        Args:
 28            serial_port (str, optional): The serial port to connect to. If not specified, it searches for available devices.
 29            baudrate (int, optional): The baud rate for the serial connection.
 30            *args: Variable length arguments to pass to the Flask constructor.
 31            **kwargs: Keyword arguments to pass to the Flask constructor.
 32        """
 33        logging.debug("Creating server.")
 34        super().__init__(*args, **kwargs)
 35        self.connected = False
 36        self.reactor_id: Optional[int] = None
 37        self.port = serial_port
 38        self.baudrate = baudrate
 39
 40        self.serial_connect()
 41
 42        # Routes
 43        @self.route("/")
 44        def root():
 45            """
 46            Root route handler.
 47            """
 48            return "REACTOR SERVER", 200
 49
 50        @self.route("/connect")
 51        def http_connect():
 52            """
 53            HTTP route for connecting to the reactor.
 54            """
 55            self.connect()
 56            return "OK", 200
 57
 58        @self.route("/reset")
 59        def http_reset():
 60            """
 61            HTTP route for resetting the connection.
 62            """
 63            self.reset()
 64            return "OK", 200
 65
 66        @self.route("/reboot")
 67        def reboot():
 68            """
 69            HTTP route for rebooting the server.
 70            """
 71            os.system("sudo reboot")
 72            return "OK", 200
 73
 74        @self.route("/send", methods=['POST'])
 75        def http_send():
 76            """
 77            HTTP route for sending commands to the reactor.
 78
 79            Returns:
 80                str: The response received from the reactor.
 81            """
 82            content = request.json
 83            logging.info(f"Received request: {content['command']}")
 84            if content['await_response']:
 85                response = self.send(content["command"])
 86            else:
 87                response = self._send(content["command"])
 88            return jsonify({"response": response}), 200
 89
 90        @self.route("/ping")
 91        def ping():
 92            """
 93            HTTP route for pinging the reactor.
 94
 95            Returns:
 96                dict: A JSON object containing reactor information (id, serial_port, hostname).
 97            """
 98            digit_regex = r"(\d+)(?!.*\d)"
 99            hostname = socket.gethostname()
100            digits = int(re.findall(digit_regex, hostname)[0])
101            self.reactor_id = digits
102            return jsonify({"id": self.reactor_id, "serial_port": self.port, "hostname": os.uname().nodename})
103
104    def __delete__(self, _):
105        self.serial.__del__()
106
107    
108    def serial_connect(self):
109        if self.port is None:
110            logging.debug("No serial port specified. Searching for available devices.")
111            self.available_ports = list_ports.comports()
112            self.available_ports = list(filter(lambda x: (x.vid,x.pid) in {(1027,24577),(9025,16),(6790,29987)},self.available_ports))
113            self.port = self.available_ports[0].device
114        self.serial = Serial(self.port, baudrate=self.baudrate, timeout=TIMEOUT)
115        logging.info(f"Connected to serial port {self.serial}.")
116
117    def connect(self):
118        """
119        Begins the connection to the reactor.
120
121        Args:
122            delay (float, optional): Delay in seconds before sending the initial command.
123        """
124        sleep(HEADER_DELAY)
125        self.serial.flush()
126        self.serial.reset_input_buffer()
127        self._send("quiet_connect")
128        self.connected = True
129
130    def reset(self):
131        """
132        Resets the connection to the reactor.
133        """
134        self.serial.flush()
135        self.serial.close()
136        self.serial.open()
137        self.connected = True
138
139    def close(self):
140        """
141        Interrupts the connection with the reactor.
142        """
143        if self.serial.is_open:
144            self.serial.reset_input_buffer()
145            self.send("fim")
146            self.serial.close()
147
148    def send(self, msg: str) -> str:
149        """
150        Sends a command to the reactor and receives the response.
151
152        Args:
153            msg (str): The command to send to the reactor.
154
155        Returns:
156            str: The response received from the reactor.
157        """
158        if not self.connected:
159            self.connect()
160        self.serial.reset_input_buffer()
161        self._send(msg)
162        return self._recv()
163
164    def _send(self, msg: str):
165        """
166        Sends a command to the reactor.
167
168        Args:
169            msg (str): The command to send to the reactor.
170        """
171        self.serial.write(msg.encode('ascii') + b'\n\r')
172
173    def _recv(self) -> str:
174        """
175        Reads from the serial port until it finds an End-of-Text ASCII token.
176
177        Returns:
178            str: The response received from the reactor.
179        """
180        response = self.serial.read_until(b'\x04') \
181            .decode('ascii') \
182            .strip("\n") \
183            .strip("\r") \
184            .strip("\x04")
185        return response
186
187    
188    def __repr__(self):
189        return f"{bcolors.OKCYAN}<Reactor at {self.port}>{bcolors.ENDC}"
190
191    def set(self, data: Optional[Dict[str, Any]] = None, **kwargs: Any):
192        """
193        Sets the value of variables based on a dictionary.
194
195        Args:
196            data (dict, optional): Dictionary containing variable-value pairs.
197            **kwargs: Additional variable-value pairs.
198
199        Examples:
200            >>> reator.set({"440": 50, "brilho": 100})
201        """
202        data = {**(data or {}), **kwargs}
203        args = ",".join(f'{k},{v}' for k, v in data.items())
204        cmd = f"set({args})"
205        self._send(cmd)
206
207    def get(self, key: Optional[str] = None) -> Any:
208        """
209        Returns the value of a variable or variables.
210
211        Args:
212            key (str, optional): The key of the variable to retrieve. If not specified, returns all variables.
213
214        Returns:
215            Any: The value of the variable(s).
216        """
217        if key is None:
218            return self._get_all()
219        if isinstance(key, str):
220            key = [key]
221        return
222
223    def _get_all(self) -> Any:
224        """
225        Returns the values of all variables.
226
227        Returns:
228            Any: The values of all variables.
229        """
230        resp = self.send("get")
231        # Parse the response and return the values of all variables.
232        return resp
233
234if __name__=="__main__":
235    rs = ReactorServer(import_name="Pyduino Slave Server")
236    rs.run(port=5000,host="0.0.0.0")