diff options
| -rwxr-xr-x | client.py | 65 | ||||
| -rwxr-xr-x | readme.rst | 24 | ||||
| -rwxr-xr-x | socketserver.py | 33 |
3 files changed, 85 insertions, 37 deletions
@@ -1,30 +1,64 @@ +#!/bin/python3
+
import sys
-from threading import Thread
from queue import Queue
-import time
+import json
-#
-# This code provide an example for connecting to the socket server.
-#
+import argparse
+from os import path
from zope import component
-
from interfaces import endpoint
+import configparser
+
+script_path = path.dirname(path.realpath(__file__))
+config_file = path.join(script_path, "config.ini")
+parser = argparse.ArgumentParser(
+ prog='client',
+ description='Command line client to the keyboard',
+ epilog='')
+parser.add_argument('configuration', nargs='?', default=config_file)
+parser.add_argument('--layer')
+args = parser.parse_args()
+
+
+config = configparser.ConfigParser(delimiters="=")
+config.read(args.configuration)
+
+if config.has_section("connection.socket"):
+ print("Socket connexion")
+ from socket_conn import SocketConnection
+ conn = SocketConnection(config["connection.socket"])
+elif config.has_section("connection.serial"):
+ print("Serial connecion")
+ from serial_conn import SerialConnection
+ conn = SerialConnection(config["connection.serial"])
+
+# Connect to the endpoint right now
component.provideAdapter(endpoint.EndPoint)
+s = component.queryAdapter(conn, endpoint.IEndpoint)
+s.connect()
-from socket_conn import SocketConnection
-conn = SocketConnection()
+if args.layer is not None:
+ print(args.layer)
+ s.queue = Queue()
+ with open(args.layer, "r") as json_file:
+ json_data = json_file.read()
+ j = json.loads(json_data)
+ content = json.dumps(j)
+ s.send(j)
+ sys.exit(0)
+from threading import Thread
+import time
class Conn(Thread):
def __init__(self, queue):
Thread.__init__(self)
self.queue = queue
- self.s = component.queryAdapter(conn, endpoint.IEndpoint)
- self.s.connect()
self.in_queue = Queue()
- self.s.queue = self.in_queue
+ s.queue = self.in_queue
def run(self):
@@ -35,15 +69,16 @@ class Conn(Thread): sys.exit(0)
elif text == "":
continue
- self.s.send([{"layout": text}])
+ j = json.loads(text)
+ s.send(j)
- if not self.s.isConnected():
+ if not s.isConnected():
time.sleep(2)
- self.s.connect()
+ s.connect()
continue
- self.s.fetch()
+ s.fetch()
while not self.in_queue.empty():
msg = self.in_queue.get(False)
@@ -109,7 +109,8 @@ the keymap can be: :a string:
- The key name will be sent: :json:`"A"`
+ The key named will be sent: :json:`"A"` (you have to follow the named
+ declared in the device)
:a list:
@@ -130,11 +131,6 @@ the keymap can be: null, null, null, null
]}
-CircuitPython provide a native library in order `to read or store json`_, and
-firmware build upon it (like KMK) are easer to use with.
-
-.. _`to read or store json`: https://docs.circuitpython.org/en/latest/docs/library/json.html
-
.. Reading message
.. ---------------
..
@@ -153,6 +149,22 @@ You can relay the events to another one instance using the network. I'm using this when I'm connected over VNC in order to use the keyboard as if it was
plugged directly in the host.
+
+----------
+The client
+----------
+
+There is a command-line client, which allow to send a layer directly to the device.
+
+The client can work in interractive mode (usefull for testing), or in a
+one-shot action, sending the json file given with the parameter :literal:`--layer`
+
----------
The device
----------
+
+CircuitPython provide a native library in order `to read or store json`_, and
+firmware build upon it (like KMK) are easer to use with.
+
+.. _`to read or store json`: https://docs.circuitpython.org/en/latest/docs/library/json.html
+
diff --git a/socketserver.py b/socketserver.py index fb71d6d..471b8be 100755 --- a/socketserver.py +++ b/socketserver.py @@ -9,6 +9,7 @@ import socket
import selectors
+import json
from dataclasses import dataclass
from typing import Callable
@@ -52,7 +53,9 @@ class Handler(object): c = waitEvent()
c.callback = self.accept
self.sel.register(self.socket, selectors.EVENT_READ, c)
- self.application_name = configuration["name"].lower()
+ self.application_name = configuration.get("name", None)
+ if self.application_name is not None:
+ self.application_name = self.application_name.lower()
self.connexions = []
self.layout_queue = layout_queue
@@ -106,23 +109,21 @@ class Handler(object): # A socket ready but sending garbage is a dead socket.
self.close(conn)
return
- last_layout = str(data, "utf-8")
+ json_data = str(data, "utf-8")
try:
- js = json.loads(last_layout)[-1]
- for key, value in js.items():
- last_layout = actions[key](value)
- except:
- print(last_layout)
-
- #title = component.queryUtility(desktopEvent.IDesktop).getForegroundWindowTitle()
- # The application name is hardcoded in the code, and should be reported
- # in the configuration or somewhere else.
- # The name of the current application is not reliable because the
- # message can be send with some lattency.
- component.handle(Debug("Received %s from the socket" % (last_layout)))
+ js = json.loads(json_data)
+ for key in js.keys():
+ component.handle(Debug("Received %s from the socket" % (key)))
+ except Exception as e:
+ print("Can’t read", json_data, e)
+ return
- # Associate the layout with the current window
- self.layout_queue.put((last_layout, self.application_name))
+ if self.application_name is not None:
+ title = self.application_name
+ else:
+ # Associate the layout with the current window
+ title = component.queryUtility(desktopEvent.IDesktop).getForegroundWindowTitle()
+ self.layout_queue.put((js, title))
def _send(self:object, conn:socket, mask:int, text) -> None:
""" Internal method used to dispatch the message to the socket. """
|