Initial re-upload of spice2x-24-08-24

This commit is contained in:
2024-08-28 11:10:34 -04:00
commit caa9e02285
1181 changed files with 380065 additions and 0 deletions

116
api/resources/python/.gitignore vendored Normal file
View File

@@ -0,0 +1,116 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/

View File

@@ -0,0 +1,24 @@
This is free and unencumbered software released into the public domain.
Anyone is free to copy, modify, publish, use, compile, sell, or
distribute this software, either in source code form or as a compiled
binary, for any purpose, commercial or non-commercial, and by any
means.
In jurisdictions that recognize copyright laws, the author or authors
of this software dedicate any and all copyright interest in the
software to the public domain. We make this dedication for the benefit
of the public at large and to the detriment of our heirs and
successors. We intend this dedication to be an overt act of
relinquishment in perpetuity of all present and future rights to this
software under copyright law.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.
For more information, please refer to <http://unlicense.org/>

View File

@@ -0,0 +1,14 @@
from .connection import Connection
from .request import Request
from .analogs import *
from .buttons import *
from .card import *
from .coin import *
from .control import *
from .exceptions import *
from .iidx import *
from .info import *
from .keypads import *
from .lights import *
from .memory import *
from .touch import *

View File

@@ -0,0 +1,28 @@
from .connection import Connection
from .request import Request
def analogs_read(con: Connection):
res = con.request(Request("analogs", "read"))
return res.get_data()
def analogs_write(con: Connection, analog_state_list):
req = Request("analogs", "write")
for state in analog_state_list:
req.add_param(state)
con.request(req)
def analogs_write_reset(con: Connection, analog_names=None):
req = Request("analogs", "write_reset")
# reset all analogs
if not analog_names:
con.request(req)
return
# reset specified analogs
for analog_name in analog_names:
req.add_param(analog_name)
con.request(req)

View File

@@ -0,0 +1,28 @@
from .connection import Connection
from .request import Request
def buttons_read(con: Connection):
res = con.request(Request("buttons", "read"))
return res.get_data()
def buttons_write(con: Connection, button_state_list):
req = Request("buttons", "write")
for state in button_state_list:
req.add_param(state)
con.request(req)
def buttons_write_reset(con: Connection, button_names=None):
req = Request("buttons", "write_reset")
# reset all buttons
if not button_names:
con.request(req)
return
# reset specified buttons
for button_name in button_names:
req.add_param(button_name)
con.request(req)

View File

@@ -0,0 +1,9 @@
from .connection import Connection
from .request import Request
def card_insert(con: Connection, unit: int, card_id: str):
req = Request("card", "insert")
req.add_param(unit)
req.add_param(card_id)
con.request(req)

View File

@@ -0,0 +1,20 @@
from .connection import Connection
from .request import Request
def coin_get(con: Connection):
res = con.request(Request("coin", "get"))
return res.get_data()[0]
def coin_set(con: Connection, amount: int):
req = Request("coin", "set")
req.add_param(amount)
con.request(req)
def coin_insert(con: Connection, amount=1):
req = Request("coin", "insert")
if amount != 1:
req.add_param(amount)
con.request(req)

View File

@@ -0,0 +1,139 @@
import os
import socket
from .request import Request
from .response import Response
from .rc4 import rc4
from .exceptions import MalformedRequestException, APIError
class Connection:
""" Container for managing a single connection to the API server.
"""
def __init__(self, host: str, port: int, password: str):
"""Default constructor.
:param host: the host string to connect to
:param port: the port of the host
:param password: the connection password string
"""
self.host = host
self.port = port
self.password = password
self.socket = None
self.cipher = None
self.reconnect()
def reconnect(self, refresh_session=True):
"""Reconnect to the server.
This opens a new connection and closes the previous one, if existing.
"""
# close old socket
self.close()
# create new socket
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.settimeout(3)
self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
self.socket.connect((self.host, self.port))
# cipher
self.change_password(self.password)
# refresh session
if refresh_session:
from .control import control_session_refresh
control_session_refresh(self)
def change_password(self, password):
"""Allows to change the password on the fly.
The cipher will be rebuilt.
"""
if len(password) > 0:
self.cipher = rc4(password.encode("UTF-8"))
else:
self.cipher = None
def close(self):
"""Close the active connection, if existing."""
# check if socket is existing
if self.socket:
# close and delete socket
self.socket.close()
self.socket = None
def request(self, request: Request):
"""Send a request to the server and receive the answer.
:param request: request object
:return: response object
"""
# check if disconnected
if not self.socket:
raise RuntimeError("No active connection.")
# build data
data = request.to_json().encode("UTF-8") + b"\x00"
if self.cipher:
data_list = list(data)
data_cipher = []
for b in data_list:
data_cipher.append(b ^ next(self.cipher))
data = bytes(data_cipher)
# send request
if os.name != 'nt':
self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK, 1)
self.socket.send(data)
# get answer
answer_data = []
while not len(answer_data) or answer_data[-1] != 0:
# receive data
if os.name != 'nt':
self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK, 1)
receive_data = self.socket.recv(4096)
# check length
if len(receive_data):
# check cipher
if self.cipher:
# add decrypted data
for b in receive_data:
answer_data.append(int(b ^ next(self.cipher)))
else:
# add plaintext
for b in receive_data:
answer_data.append(int(b))
else:
raise RuntimeError("Connection was closed.")
# check for empty response
if len(answer_data) <= 1:
# empty response means the JSON couldn't be parsed
raise MalformedRequestException()
# build response
response = Response(bytes(answer_data[:-1]).decode("UTF-8"))
if len(response.get_errors()):
raise APIError(response.get_errors())
# check ID
req_id = request.get_id()
res_id = response.get_id()
if req_id != res_id:
raise RuntimeError(f"Unexpected response ID: {res_id} (expected {req_id})")
# return response object
return response

View File

@@ -0,0 +1,51 @@
import random
from .connection import Connection
from .request import Request
def control_raise(con: Connection, signal: str):
req = Request("control", "raise")
req.add_param(signal)
con.request(req)
def control_exit(con: Connection, code=None):
req = Request("control", "exit")
if code:
req.add_param(code)
try:
con.request(req)
except RuntimeError:
pass # we expect the connection to get killed
def control_restart(con: Connection):
req = Request("control", "restart")
try:
con.request(req)
except RuntimeError:
pass # we expect the connection to get killed
def control_session_refresh(con: Connection):
res = con.request(Request("control", "session_refresh", req_id=random.randint(1, 2**64)))
# apply new password
password = res.get_data()[0]
con.change_password(password)
def control_shutdown(con: Connection):
req = Request("control", "shutdown")
try:
con.request(req)
except RuntimeError:
pass # we expect the connection to get killed
def control_reboot(con: Connection):
req = Request("control", "reboot")
try:
con.request(req)
except RuntimeError:
pass # we expect the connection to get killed

View File

@@ -0,0 +1,10 @@
class APIError(Exception):
def __init__(self, errors):
super().__init__("\r\n".join(errors))
class MalformedRequestException(Exception):
pass

View File

@@ -0,0 +1,18 @@
from .connection import Connection
from .request import Request
def iidx_ticker_get(con: Connection):
res = con.request(Request("iidx", "ticker_get"))
return res.get_data()
def iidx_ticker_set(con: Connection, text: str):
req = Request("iidx", "ticker_set")
req.add_param(text)
con.request(req)
def iidx_ticker_reset(con: Connection):
req = Request("iidx", "ticker_reset")
con.request(req)

View File

@@ -0,0 +1,17 @@
from .connection import Connection
from .request import Request
def info_avs(con: Connection):
res = con.request(Request("info", "avs"))
return res.get_data()[0]
def info_launcher(con: Connection):
res = con.request(Request("info", "launcher"))
return res.get_data()[0]
def info_memory(con: Connection):
res = con.request(Request("info", "memory"))
return res.get_data()[0]

View File

@@ -0,0 +1,24 @@
from .connection import Connection
from .request import Request
def keypads_write(con: Connection, keypad: int, input_values: str):
req = Request("keypads", "write")
req.add_param(keypad)
req.add_param(input_values)
con.request(req)
def keypads_set(con: Connection, keypad: int, input_values: str):
req = Request("keypads", "set")
req.add_param(keypad)
for value in input_values:
req.add_param(value)
con.request(req)
def keypads_get(con: Connection, keypad: int):
req = Request("keypads", "get")
req.add_param(keypad)
res = con.request(req)
return res.get_data()

View File

@@ -0,0 +1,28 @@
from .connection import Connection
from .request import Request
def lights_read(con: Connection):
res = con.request(Request("lights", "read"))
return res.get_data()
def lights_write(con: Connection, light_state_list):
req = Request("lights", "write")
for state in light_state_list:
req.add_param(state)
con.request(req)
def lights_write_reset(con: Connection, light_names=None):
req = Request("lights", "write_reset")
# reset all lights
if not light_names:
con.request(req)
return
# reset specified lights
for light_name in light_names:
req.add_param(light_name)
con.request(req)

View File

@@ -0,0 +1,31 @@
from .connection import Connection
from .request import Request
def memory_write(con: Connection, dll_name: str, data: str, offset: int):
req = Request("memory", "write")
req.add_param(dll_name)
req.add_param(data)
req.add_param(offset)
con.request(req)
def memory_read(con: Connection, dll_name: str, offset: int, size: int):
req = Request("memory", "read")
req.add_param(dll_name)
req.add_param(offset)
req.add_param(size)
res = con.request(req)
return res.get_data()[0]
def memory_signature(con: Connection, dll_name: str, signature: str,
replacement: str, offset: int, usage: int):
req = Request("memory", "signature")
req.add_param(dll_name)
req.add_param(signature)
req.add_param(replacement)
req.add_param(offset)
req.add_param(usage)
res = con.request(req)
return res.get_data()[0]

View File

@@ -0,0 +1,24 @@
def rc4_ksa(key):
n = len(key)
j = 0
s_box = list(range(256))
for i in range(256):
j = (j + s_box[i] + key[i % n]) % 256
s_box[i], s_box[j] = s_box[j], s_box[i]
return s_box
def rc4_prga(s_box):
i = 0
j = 0
while True:
i = (i + 1) % 256
j = (j + s_box[i]) % 256
s_box[i], s_box[j] = s_box[j], s_box[i]
yield s_box[(s_box[i] + s_box[j]) % 256]
def rc4(key):
return rc4_prga(rc4_ksa(key))

View File

@@ -0,0 +1,63 @@
import json
from threading import Lock
class Request:
# global ID pool
GLOBAL_ID = 1
GLOBAL_ID_LOCK = Lock()
def __init__(self, module: str, function: str, req_id=None):
# use global ID
with Request.GLOBAL_ID_LOCK:
if req_id is None:
# reset at max value
Request.GLOBAL_ID += 1
if Request.GLOBAL_ID >= 2 ** 64:
Request.GLOBAL_ID = 1
# get ID and increase by one
req_id = Request.GLOBAL_ID
else:
# carry over ID
Request.GLOBAL_ID = req_id
# remember ID
self._id = req_id
# build data dict
self.data = {
"id": req_id,
"module": module,
"function": function,
"params": []
}
@staticmethod
def from_json(request_json: str):
req = Request("", "", 0)
req.data = json.loads(request_json)
req._id = req.data["id"]
return req
def get_id(self):
return self._id
def to_json(self):
return json.dumps(
self.data,
ensure_ascii=False,
check_circular=False,
allow_nan=False,
indent=None,
separators=(",", ":"),
sort_keys=False
)
def add_param(self, param):
self.data["params"].append(param)

View File

@@ -0,0 +1,30 @@
import json
class Response:
def __init__(self, response_json: str):
self._res = json.loads(response_json)
self._id = self._res["id"]
self._errors = self._res["errors"]
self._data = self._res["data"]
def to_json(self):
return json.dumps(
self._res,
ensure_ascii=True,
check_circular=False,
allow_nan=False,
indent=2,
separators=(",", ": "),
sort_keys=False
)
def get_id(self):
return self._id
def get_errors(self):
return self._errors
def get_data(self):
return self._data

View File

@@ -0,0 +1,21 @@
from .connection import Connection
from .request import Request
def touch_read(con: Connection):
res = con.request(Request("touch", "read"))
return res.get_data()
def touch_write(con: Connection, touch_points):
req = Request("touch", "write")
for state in touch_points:
req.add_param(state)
con.request(req)
def touch_write_reset(con: Connection, touch_ids):
req = Request("touch", "write_reset")
for touch_id in touch_ids:
req.add_param(touch_id)
con.request(req)

View File

@@ -0,0 +1,574 @@
#!/usr/bin/env python3
from datetime import datetime
import tkinter as tk
import tkinter.ttk as ttk
import tkinter.messagebox
try:
import spiceapi
except ModuleNotFoundError:
raise RuntimeError("spiceapi module not installed")
NSEW = tk.N+tk.S+tk.E+tk.W
def api_action(func):
def wrapper(*args, **kwargs):
try:
func(*args, **kwargs)
except spiceapi.APIError as e:
tk.messagebox.showerror(title="API Error", message=str(e))
except ValueError as e:
tk.messagebox.showerror(title="Input Error", message=str(e))
except (ConnectionResetError, BrokenPipeError, RuntimeError) as e:
tk.messagebox.showerror(title="Connection Error", message=str(e))
except AttributeError:
tk.messagebox.showerror(title="Error", message="No active connection.")
except BaseException as e:
tk.messagebox.showerror(title="Exception", message=str(e))
return wrapper
class TextField(ttk.Frame):
"""Simple text field with a scroll bar to the right."""
def __init__(self, parent, read_only=False, **kwargs):
# init frame
ttk.Frame.__init__(self, parent, **kwargs)
self.parent = parent
self.read_only = read_only
# create scroll bar
self.scroll = ttk.Scrollbar(self)
self.scroll.pack(side=tk.RIGHT, fill=tk.Y)
# create text field
self.contents = tk.Text(self, height=1, width=50,
foreground="black", background="#E8E6E0",
insertbackground="black")
self.contents.pack(side=tk.LEFT, expand=True, fill=tk.BOTH)
# link scroll bar with text field
self.scroll.config(command=self.contents.yview)
self.contents.config(yscrollcommand=self.scroll.set)
# read only setting
if self.read_only:
self.contents.config(state=tk.DISABLED)
def get_text(self):
"""Get the current text content as string.
:return: text content
"""
return self.contents.get("1.0", tk.END+"-1c")
def set_text(self, text):
"""Set the current text content.
:param text: text content
:return: None
"""
# enable if read only
if self.read_only:
self.contents.config(state=tk.NORMAL)
# delete old content and insert replacement text
self.contents.delete("1.0", tk.END)
self.contents.insert(tk.END, text)
# disable if read only
if self.read_only:
self.contents.config(state=tk.DISABLED)
class ManualTab(ttk.Frame):
"""Manual JSON request/response functinality."""
def __init__(self, app, parent, **kwargs):
# init frame
ttk.Frame.__init__(self, parent, **kwargs)
self.app = app
self.parent = parent
self.rowconfigure(0, weight=1)
self.rowconfigure(1, weight=1)
self.columnconfigure(0, weight=1)
# request text field
self.txt_request = TextField(self, read_only=False)
self.txt_request.grid(row=0, column=0, sticky=NSEW, padx=2, pady=2)
self.txt_request.set_text(
'{\n'
' "id": 1,\n'
' "module": "coin",\n'
' "function": "insert",\n'
' "params": []\n'
'}\n'
)
# response text field
self.txt_response = TextField(self, read_only=False)
self.txt_response.grid(row=1, column=0, sticky=NSEW, padx=2, pady=2)
#self.txt_response.contents.config(state=tk.DISABLED)
# send button
self.btn_send = ttk.Button(self, text="Send", command=self.action_send)
self.btn_send.grid(row=2, column=0, sticky=tk.W+tk.E, padx=2, pady=2)
def action_send(self):
"""Gets called when the send button is pressed."""
# check connection
if self.app.connection:
# send request and get response
self.txt_response.set_text("Sending...")
try:
# build request
request = spiceapi.Request.from_json(self.txt_request.get_text())
# send request and get response, measure time
t1 = datetime.now()
response = self.app.connection.request(request)
t2 = datetime.now()
# set response text
self.txt_response.set_text("{}\n\nElapsed time: {} seconds".format(
response.to_json(),
(t2 - t1).total_seconds())
)
except spiceapi.APIError as e:
self.txt_response.set_text(f"Server returned error:\n{e}")
except spiceapi.MalformedRequestException:
self.txt_response.set_text("Malformed request detected.")
except (ConnectionResetError, BrokenPipeError, RuntimeError) as e:
self.txt_response.set_text("Error sending request: " + str(e))
except BaseException as e:
self.txt_response.set_text("General Exception: " + str(e))
else:
# print error
self.txt_response.set_text("No active connection.")
class ControlTab(ttk.Frame):
"""Main control tab."""
def __init__(self, app, parent, **kwargs):
# init frame
ttk.Frame.__init__(self, parent, **kwargs)
self.app = app
self.parent = parent
# scale grid
self.columnconfigure(0, weight=1)
# card
self.card = ttk.Frame(self, padding=(8, 8, 8, 8))
self.card.grid(row=0, column=0, sticky=tk.E+tk.W)
self.card.columnconfigure(0, weight=1)
self.card.columnconfigure(1, weight=1)
self.card_lbl = ttk.Label(self.card, text="Card")
self.card_lbl.grid(row=0, columnspan=2)
self.card_entry = ttk.Entry(self.card)
self.card_entry.insert(tk.END, "E004010000000000")
self.card_entry.grid(row=1, columnspan=2, sticky=NSEW, padx=2, pady=2)
self.card_insert_p1 = ttk.Button(self.card, text="Insert P1", command=self.action_insert_p1)
self.card_insert_p1.grid(row=2, column=0, sticky=NSEW, padx=2, pady=2)
self.card_insert_p2 = ttk.Button(self.card, text="Insert P2", command=self.action_insert_p2)
self.card_insert_p2.grid(row=2, column=1, sticky=NSEW, padx=2, pady=2)
# coin
self.coin = ttk.Frame(self, padding=(8, 8, 8, 8))
self.coin.grid(row=1, column=0, sticky=tk.E+tk.W)
self.coin.columnconfigure(0, weight=1)
self.coin.columnconfigure(1, weight=1)
self.coin.columnconfigure(2, weight=1)
self.coin_lbl = ttk.Label(self.coin, text="Coins")
self.coin_lbl.grid(row=0, columnspan=3)
self.coin_entry = ttk.Entry(self.coin)
self.coin_entry.insert(tk.END, "1")
self.coin_entry.grid(row=1, columnspan=3, sticky=NSEW, padx=2, pady=2)
self.coin_set = ttk.Button(self.coin, text="Set to Amount", command=self.action_coin_set)
self.coin_set.grid(row=2, column=0, sticky=NSEW, padx=2, pady=2)
self.coin_insert = ttk.Button(self.coin, text="Insert Amount", command=self.action_coin_insert)
self.coin_insert.grid(row=2, column=1, sticky=NSEW, padx=2, pady=2)
self.coin_insert = ttk.Button(self.coin, text="Insert Single", command=self.action_coin_insert_single)
self.coin_insert.grid(row=2, column=2, sticky=NSEW, padx=2, pady=2)
@api_action
def action_insert(self, unit: int):
spiceapi.card_insert(self.app.connection, unit, self.card_entry.get())
@api_action
def action_insert_p1(self):
return self.action_insert(0)
@api_action
def action_insert_p2(self):
return self.action_insert(1)
@api_action
def action_coin_set(self):
spiceapi.coin_set(self.app.connection, int(self.coin_entry.get()))
@api_action
def action_coin_insert(self):
spiceapi.coin_insert(self.app.connection, int(self.coin_entry.get()))
@api_action
def action_coin_insert_single(self):
spiceapi.coin_insert(self.app.connection)
class InfoTab(ttk.Frame):
"""The info tab."""
def __init__(self, app, parent, **kwargs):
# init frame
ttk.Frame.__init__(self, parent, **kwargs)
self.app = app
self.parent = parent
self.rowconfigure(0, weight=1)
self.columnconfigure(0, weight=1)
# info text field
self.txt_info = TextField(self, read_only=True)
self.txt_info.grid(row=0, column=0, sticky=NSEW, padx=2, pady=2)
# refresh button
self.btn_refresh = ttk.Button(self, text="Refresh", command=self.action_refresh)
self.btn_refresh.grid(row=1, column=0, sticky=tk.W+tk.E, padx=2, pady=2)
@api_action
def action_refresh(self):
# get information
avs = spiceapi.info_avs(self.app.connection)
launcher = spiceapi.info_launcher(self.app.connection)
memory = spiceapi.info_memory(self.app.connection)
# build text
avs_text = ""
for k, v in avs.items():
avs_text += f"{k}: {v}\n"
launcher_text = ""
for k, v in launcher.items():
if isinstance(v, list):
launcher_text += f"{k}:\n"
for i in v:
launcher_text += f" {i}\n"
else:
launcher_text += f"{k}: {v}\n"
memory_text = ""
for k, v in memory.items():
memory_text += f"{k}: {v}\n"
# set text
self.txt_info.set_text(
f"AVS:\n{avs_text}\n"
f"Launcher:\n{launcher_text}\n"
f"Memory:\n{memory_text}"
)
class ButtonsTab(ttk.Frame):
"""The buttons tab."""
def __init__(self, app, parent, **kwargs):
# init frame
ttk.Frame.__init__(self, parent, **kwargs)
self.app = app
self.parent = parent
self.rowconfigure(0, weight=1)
self.columnconfigure(0, weight=1)
# button text field
self.txt_buttons = TextField(self, read_only=True)
self.txt_buttons.grid(row=0, column=0, sticky=NSEW, padx=2, pady=2)
# refresh button
self.btn_refresh = ttk.Button(self, text="Refresh", command=self.action_refresh)
self.btn_refresh.grid(row=1, column=0, sticky=tk.W+tk.E, padx=2, pady=2)
@api_action
def action_refresh(self):
# get states
states = spiceapi.buttons_read(self.app.connection)
# build text
txt = ""
for name, velocity, active in states:
state = "on" if velocity > 0 else "off"
active_txt = "" if active else " (inactive)"
txt += f"{name}: {state}{active_txt}\n"
if len(states) == 0:
txt = "No buttons available."
# set text
self.txt_buttons.set_text(txt)
class AnalogsTab(ttk.Frame):
"""The analogs tab."""
def __init__(self, app, parent, **kwargs):
# init frame
ttk.Frame.__init__(self, parent, **kwargs)
self.app = app
self.parent = parent
self.rowconfigure(0, weight=1)
self.columnconfigure(0, weight=1)
# button text field
self.txt_analogs = TextField(self, read_only=True)
self.txt_analogs.grid(row=0, column=0, sticky=NSEW, padx=2, pady=2)
# refresh button
self.btn_refresh = ttk.Button(self, text="Refresh", command=self.action_refresh)
self.btn_refresh.grid(row=1, column=0, sticky=tk.W+tk.E, padx=2, pady=2)
@api_action
def action_refresh(self):
# get states
states = spiceapi.analogs_read(self.app.connection)
# build text
txt = ""
for name, value, active in states:
value_txt = round(value, 2)
active_txt = "" if active else " (inactive)"
txt += f"{name}: {value_txt}{active_txt}\n"
if len(states) == 0:
txt = "No analogs available."
# set text
self.txt_analogs.set_text(txt)
class LightsTab(ttk.Frame):
"""The lights tab."""
def __init__(self, app, parent, **kwargs):
# init frame
ttk.Frame.__init__(self, parent, **kwargs)
self.app = app
self.parent = parent
self.rowconfigure(0, weight=1)
self.columnconfigure(0, weight=1)
# light text field
self.txt_lights = TextField(self, read_only=True)
self.txt_lights.grid(row=0, column=0, sticky=NSEW, padx=2, pady=2)
# refresh button
self.btn_refresh = ttk.Button(self, text="Refresh", command=self.action_refresh)
self.btn_refresh.grid(row=1, column=0, sticky=tk.W+tk.E, padx=2, pady=2)
@api_action
def action_refresh(self):
# get states
states = spiceapi.lights_read(self.app.connection)
# build text
txt = ""
for name, value, active in states:
value_txt = round(value, 2)
active_txt = "" if active else " (inactive)"
txt += f"{name}: {value_txt}{active_txt}\n"
if len(states) == 0:
txt = "No lights available."
# set text
self.txt_lights.set_text(txt)
class MainApp(ttk.Frame):
"""The main application frame."""
def __init__(self, parent, **kwargs):
# init frame
ttk.Frame.__init__(self, parent, **kwargs)
self.parent = parent
self.connection = None
self.tabs = ttk.Notebook(self)
self.tab_control = ControlTab(self, self.tabs)
self.tabs.add(self.tab_control, text="Control")
self.tab_info = InfoTab(self, self.tabs)
self.tabs.add(self.tab_info, text="Info")
self.tab_buttons = ButtonsTab(self, self.tabs)
self.tabs.add(self.tab_buttons, text="Buttons")
self.tab_analogs = AnalogsTab(self, self.tabs)
self.tabs.add(self.tab_analogs, text="Analogs")
self.tab_lights = LightsTab(self, self.tabs)
self.tabs.add(self.tab_lights, text="Lights")
self.tab_manual = ManualTab(self, self.tabs)
self.tabs.add(self.tab_manual, text="Manual")
self.tabs.pack(expand=True, fill=tk.BOTH)
# connection panel
self.frm_connection = ttk.Frame(self)
# host: [field]
self.lbl_host = ttk.Label(self.frm_connection, text="Host:")
self.txt_host = ttk.Entry(self.frm_connection, width=15)
self.lbl_host.columnconfigure(0, weight=1)
self.txt_host.columnconfigure(1, weight=1)
self.txt_host.insert(tk.END, "localhost")
# port: [field]
self.lbl_port = ttk.Label(self.frm_connection, text="Port:")
self.txt_port = ttk.Entry(self.frm_connection, width=5)
self.lbl_port.columnconfigure(2, weight=1)
self.txt_port.columnconfigure(3, weight=1)
self.txt_port.insert(tk.END, "1337")
# pass: [field]
self.lbl_pw = ttk.Label(self.frm_connection, text="Pass:")
self.txt_pw = ttk.Entry(self.frm_connection, width=10)
self.lbl_pw.columnconfigure(4, weight=1)
self.txt_pw.columnconfigure(5, weight=1)
self.txt_pw.insert(tk.END, "debug")
# grid setup
self.lbl_host.grid(row=0, column=0, sticky=tk.W+tk.E, padx=2)
self.txt_host.grid(row=0, column=1, sticky=tk.W+tk.E, padx=2)
self.lbl_port.grid(row=0, column=2, sticky=tk.W+tk.E, padx=2)
self.txt_port.grid(row=0, column=3, sticky=tk.W+tk.E, padx=2)
self.lbl_pw.grid(row=0, column=4, sticky=tk.W+tk.E, padx=2)
self.txt_pw.grid(row=0, column=5, sticky=tk.W+tk.E, padx=2)
self.frm_connection.pack(fill=tk.NONE, pady=2)
# send/connect/disconnect buttons panel
self.frm_connect = ttk.Frame(self)
# connect button
self.btn_connect = ttk.Button(
self.frm_connect,
text="Connect",
command=self.action_connect,
width=10)
# disconnect button
self.btn_disconnect = ttk.Button(
self.frm_connect,
text="Disconnect",
command=self.action_disconnect,
width=10)
# kill button
self.btn_kill = ttk.Button(
self.frm_connect,
text="Kill",
command=self.action_kill,
width=10)
# restart button
self.btn_restart = ttk.Button(
self.frm_connect,
text="Restart",
command=self.action_restart,
width=10)
# grid setup
self.btn_connect.grid(row=0, column=1, sticky=tk.W+tk.E, padx=2)
self.btn_disconnect.grid(row=0, column=2, sticky=tk.W+tk.E, padx=2)
self.btn_kill.grid(row=0, column=3, sticky=tk.W+tk.E, padx=2)
self.btn_restart.grid(row=0, column=4, sticky=tk.W+tk.E, padx=2)
self.frm_connect.pack(fill=tk.NONE, pady=2)
def action_connect(self):
"""Gets called when the connect button is pressed."""
# retrieve connection info
host = self.txt_host.get()
port = self.txt_port.get()
password = self.txt_pw.get()
# check input
if not port.isdigit():
tk.messagebox.showerror("Connection Error", f"Port '{port}' is not a valid number.")
return
# create a new connection
try:
self.connection = spiceapi.Connection(host=host, port=int(port), password=password)
except OSError as e:
tk.messagebox.showerror("Connection Error", "Failed to connect: " + str(e))
return
# print success
tk.messagebox.showinfo("Success", "Connected.")
def action_disconnect(self):
"""Gets called when the disconnect button is pressed."""
# check connection
if self.connection:
# close connection
self.connection.close()
self.connection = None
tk.messagebox.showinfo("Success", "Closed connection.")
else:
# print error
tk.messagebox.showinfo("Error", "No active connection.")
def action_kill(self):
"""Gets called when the kill button is pressed."""
# check connection
if self.connection:
spiceapi.control_exit(self.connection)
self.connection = None
else:
tk.messagebox.showinfo("Error", "No active connection.")
def action_restart(self):
"""Gets called when the restart button is pressed."""
# check connection
if self.connection:
spiceapi.control_restart(self.connection)
self.connection = None
else:
tk.messagebox.showinfo("Error", "No active connection.")
if __name__ == "__main__":
# create root
root = tk.Tk()
root.title("SpiceRemote")
root.geometry("500x300")
# set theme
preferred_theme = "clam"
s = ttk.Style(root)
if preferred_theme in s.theme_names():
s.theme_use(preferred_theme)
# add application
MainApp(root).pack(side=tk.TOP, fill=tk.BOTH, expand=True)
# run root
root.mainloop()

View File

@@ -0,0 +1,52 @@
#!/usr/bin/env python3
import binascii
import spiceapi
import argparse
def patch_string(con, dll_name: str, find: str, replace: str):
while True:
try:
# replace first result
address = spiceapi.memory_signature(
con, dll_name,
binascii.hexlify(bytes(find, "utf-8")).decode("utf-8"),
binascii.hexlify(bytes(replace, "utf-8")).decode("utf-8"),
0, 0)
# print findings
print("{}: {} = {} => {}".format(
dll_name,
hex(address),
find,
replace))
except spiceapi.APIError:
# this happens when the signature wasn't found anymore
break
def main():
# parse args
parser = argparse.ArgumentParser(description="SpiceAPI string replacer")
parser.add_argument("host", type=str, help="The host to connect to")
parser.add_argument("port", type=int, help="The port the host is using")
parser.add_argument("password", type=str, help="The pass the host is using")
parser.add_argument("dll", type=str, help="The DLL to patch")
parser.add_argument("find", type=str, help="The string to find")
parser.add_argument("replace", type=str, help="The string to replace with")
args = parser.parse_args()
# connect
con = spiceapi.Connection(host=args.host, port=args.port, password=args.password)
# replace the string
patch_string(con, args.dll, args.find, args.replace)
if __name__ == "__main__":
main()