zjamnik/joycontrol
1
0
Fork 0
forked from mirror/joycontrol
Code Pull Requests Activity

Merge branch 'nfc_cleanup' into nfc-r3

Browse Source
This commit is contained in:
Robert Martin
2020-05-04 16:12:36 +02:00
parent c17ab21140 3adf0b2878
commit bab7d73cfa
6 changed files with 154 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
joycontrol/command_line_interface.py
+48 -11
View File
@@ -38,15 +38,16 @@ def _print_doc(string):
print(line[prefix_i:] if line.strip() else line)
class ControllerCLI:
def __init__(self, controller_state: ControllerState):
self.controller_state = controller_state
class CLI:
def __init__(self):
self.commands = {}
def add_command(self, name, command):
if name in self.commands:
raise ValueError(f'Command {name} already registered.')
self.commands[name] = command
async def cmd_help(self):
print('Button commands:')
print(', '.join(self.controller_state.button_state.get_available_buttons()))
print()
print('Commands:')
for name, fun in inspect.getmembers(self):
if name.startswith('cmd_') and fun.__doc__:
@@ -59,6 +60,47 @@ class ControllerCLI:
print('Commands can be chained using "&&"')
print('Type "exit" to close.')
async def run(self):
while True:
user_input = await ainput(prompt='cmd >> ')
if not user_input:
continue
for command in user_input.split('&&'):
cmd, *args = shlex.split(command)
if cmd == 'exit':
return
if hasattr(self, f'cmd_{cmd}'):
try:
result = await getattr(self, f'cmd_{cmd}')(*args)
if result:
print(result)
except Exception as e:
print(e)
elif cmd in self.commands:
try:
result = await self.commands[cmd](*args)
if result:
print(result)
except Exception as e:
print(e)
else:
print('command', cmd, 'not found, call help for help.')
class ControllerCLI(CLI):
def __init__(self, controller_state: ControllerState):
super().__init__()
self.controller_state = controller_state
async def cmd_help(self):
print('Button commands:')
print(', '.join(self.controller_state.button_state.get_available_buttons()))
print()
await super().cmd_help()
@staticmethod
def _set_stick(stick, direction, value):
if direction == 'center':
@@ -109,11 +151,6 @@ class ControllerCLI:
else:
raise ValueError('Value of side must be "l", "left" or "r", "right"')
def add_command(self, name, command):
if name in self.commands:
raise ValueError(f'Command {name} already registered.')
self.commands[name] = command
async def run(self):
while True:
user_input = await ainput(prompt='cmd >> ')
joycontrol/controller_state.py
+3
View File
@@ -51,6 +51,9 @@ class ControllerState:
def set_nfc(self, nfc_content):
self._nfc_content = nfc_content
def get_nfc(self):
return self._nfc_content
async def send(self):
"""
Invokes protocol.send_controller_state(). Returns after the controller state was send.
joycontrol/mcu.py → joycontrol/ir_nfc_mcu.py
+12 -11
View File
@@ -1,6 +1,9 @@
import logging
from enum import Enum
from crc8 import crc8
logger = logging.getLogger(__name__)
class Action(Enum):
NON = 0
@@ -24,7 +27,12 @@ def copyarray(dest, offset, src):
for i in range(len(src)):
dest[offset + i] = src[i]
class Mcu:
class IrNfcMcu:
"""
TODO: cleanup
"""
def __init__(self):
self._fw_major = [0, 3]
self._fw_minor = [0, 5]
@@ -79,14 +87,7 @@ class Mcu:
def update_nfc_report(self):
self._bytes = [0] * 313
if self.get_action() == Action.REQUEST_STATUS:
self._bytes[0] = 1
self._bytes[1] = 0
self._bytes[2] = 0
self._bytes[3] = self._fw_major[0]
self._bytes[4] = self._fw_major[1]
self._bytes[5] = self._fw_minor[0]
self._bytes[6] = self._fw_minor[1]
self._bytes[7] = self._get_state_byte()
self.update_status()
elif self.get_action() == Action.NON:
self._bytes[0] = 0xff
elif self.get_action() == Action.START_TAG_DISCOVERY:
@@ -104,13 +105,13 @@ class Mcu:
self._bytes[2] = 5
self._bytes[3] = 0
self._bytes[4] = 0
if not self._nfc_content is None:
if self._nfc_content is not None:
data = [0x09, 0x31, 0x09, 0x00, 0x00, 0x00, 0x01, 0x01, 0x02, 0x00, 0x07]
copyarray(self._bytes, 5, data)
copyarray(self._bytes, 5 + len(data), self._nfc_content[0:3])
copyarray(self._bytes, 5 + len(data) + 3, self._nfc_content[4:8])
else:
print('nfc content is none')
logger.info('nfc content is none')
self._bytes[5] = 9
self._bytes[6] = 0x31
self._bytes[7] = 0
joycontrol/protocol.py
+67 -44
View File
@@ -1,5 +1,6 @@
import asyncio
import logging
import time
from asyncio import BaseTransport, BaseProtocol
from contextlib import suppress
from typing import Optional, Union, Tuple, Text
@@ -10,7 +11,7 @@ from joycontrol.controller_state import ControllerState
from joycontrol.memory import FlashMemory
from joycontrol.report import OutputReport, SubCommand, InputReport, OutputReportID
from joycontrol.transport import NotConnectedError
from joycontrol.mcu import Mcu, McuState, Action
from joycontrol.ir_nfc_mcu import IrNfcMcu, McuState, Action
from crc8 import crc8
logger = logging.getLogger(__name__)
@@ -41,7 +42,7 @@ class ControllerProtocol(BaseProtocol):
self._controller_state = ControllerState(self, controller, spi_flash=spi_flash)
self._controller_state_sender = None
self._mcu = Mcu()
self._mcu = IrNfcMcu()
# None = Just answer to sub commands
self._input_report_mode = None
@@ -55,7 +56,7 @@ class ControllerProtocol(BaseProtocol):
Raises NotConnected exception if the transport is not connected or the connection was lost.
"""
# TODO: Call write directly if not in 0x30 input report mode
# TODO: Call write directly if in continuously sending input report mode
if self.transport is None:
raise NotConnectedError('Transport not registered.')
@@ -93,11 +94,6 @@ class ControllerProtocol(BaseProtocol):
input_report.set_timer(self._input_report_timer)
self._input_report_timer = (self._input_report_timer + 1) % 0x100
if input_report.get_input_report_id() == 0x31:
self._mcu.set_nfc(self._controller_state._nfc_content)
self._mcu.update_nfc_report()
input_report.set_mcu(self._mcu)
await self.transport.write(input_report)
self._controller_state.sig_is_send.set()
@@ -131,28 +127,29 @@ class ControllerProtocol(BaseProtocol):
async def input_report_mode_full(self):
"""
Continuously sends full input reports containing the controller state.
Continuously sends:
0x30 input reports containing the controller state OR
0x31 input reports containing the controller state and nfc data
"""
if self.transport.is_reading():
raise ValueError('Transport must be paused in full input report mode')
# send state at 66Hz
send_delay = 0.015
await asyncio.sleep(send_delay)
last_send_time = time.time()
input_report = InputReport()
input_report.set_vibrator_input()
input_report.set_misc()
if self._input_report_mode is None:
raise ValueError('Input report mode is not set.')
input_report.set_input_report_id(self._input_report_mode)
reader = asyncio.ensure_future(self.transport.read())
try:
while True:
input_report.set_input_report_id(self._input_report_mode)
# TODO: improve timing
if self.controller == Controller.PRO_CONTROLLER:
# send state at 120Hz
await asyncio.sleep(1 / 120)
else:
# send state at 60Hz
await asyncio.sleep(1 / 60)
reply_send = False
if reader.done():
data = await reader
@@ -168,8 +165,10 @@ class ControllerProtocol(BaseProtocol):
pass
elif output_report_id == OutputReportID.SUB_COMMAND:
reply_send = await self._reply_to_sub_command(report)
elif output_report_id == OutputReportID.REQUEST_MCU:
reply_send = await self._reply_to_mcu(report)
elif output_report_id == OutputReportID.REQUEST_IR_NFC_MCU:
# TODO: This does not reply anything
# reply_send = await self._reply_to_ir_nfc_mcu(report)
await self._reply_to_ir_nfc_mcu(report)
else:
logger.warning(f'Report unknown output report "{output_report_id}" - IGNORE')
except ValueError as v_err:
@@ -185,8 +184,26 @@ class ControllerProtocol(BaseProtocol):
# TODO: set some sensor data
input_report.set_6axis_data()
# set nfc data
if input_report.get_input_report_id() == 0x31:
self._mcu.set_nfc(self._controller_state.get_nfc())
self._mcu.update_nfc_report()
input_report.set_ir_nfc_data(bytes(self._mcu))
await self.write(input_report)
# calculate delay
current_time = time.time()
time_delta = time.time() - last_send_time
sleep_time = send_delay - time_delta
last_send_time = current_time
if sleep_time < 0:
# logger.warning(f'Code is running {abs(sleep_time)} s too slow!')
sleep_time = 0
await asyncio.sleep(sleep_time)
except NotConnectedError as err:
# Stop 0x30 input report mode if disconnected.
logger.error(err)
@@ -220,13 +237,17 @@ class ControllerProtocol(BaseProtocol):
else:
logger.warning(f'Output report {output_report_id} not implemented - ignoring')
async def _reply_to_mcu(self, report):
async def _reply_to_ir_nfc_mcu(self, report):
"""
TODO: Cleanup
We aren't replying to anything here, do we need to?
"""
sub_command = report.data[11]
sub_command_data = report.data[12:]
# logging.info(f'received output report - Request MCU sub command {sub_command}')
if self._mcu.get_action() == Action.READ_TAG or self._mcu.get_action() == Action.READ_TAG_2 or self._mcu.get_action() == Action.READ_FINISHED:
if self._mcu.get_action() in (Action.READ_TAG, Action.READ_TAG_2, Action.READ_FINISHED):
return
# Request mcu state
@@ -369,16 +390,35 @@ class ControllerProtocol(BaseProtocol):
await self.write(input_report)
async def _command_set_input_report_mode(self, sub_command_data):
if sub_command_data[0] == 0x30:
pass
elif sub_command_data[0] == 0x31:
pass
if self._input_report_mode == sub_command_data[0]:
logger.warning(f'Already in input report mode {sub_command_data[0]} - ignoring request')
# Start input report reader
if sub_command_data[0] in (0x30, 0x31):
new_reader = asyncio.ensure_future(self.input_report_mode_full())
else:
logger.error(f'input report mode {sub_command_data[0]} not implemented - ignoring request')
return
logger.info(f'Setting input report mode to {hex(sub_command_data[0])}...')
# Replace the currently running reader with the input report mode sender,
# which will also handle incoming requests in the future
self.transport.pause_reading()
# We need to replace the reader in the future because this function was probably called by it
async def set_reader():
await self.transport.set_reader(new_reader)
logger.info(f'Setting input report mode to {hex(sub_command_data[0])}...')
self._input_report_mode = sub_command_data[0]
self.transport.resume_reading()
asyncio.ensure_future(set_reader()).add_done_callback(
utils.create_error_check_callback()
)
# Send acknowledgement
input_report = InputReport()
input_report.set_input_report_id(0x21)
input_report.set_misc()
@@ -388,23 +428,6 @@ class ControllerProtocol(BaseProtocol):
await self.write(input_report)
# start sending input reports
if self._input_report_mode is None:
self.transport.pause_reading()
new_reader = asyncio.ensure_future(self.input_report_mode_full())
# We need to swap the reader in the future because this function was probably called by it
async def set_reader():
await self.transport.set_reader(new_reader)
self.transport.resume_reading()
asyncio.ensure_future(set_reader()).add_done_callback(
utils.create_error_check_callback()
)
self._input_report_mode = sub_command_data[0]
async def _command_trigger_buttons_elapsed_time(self, sub_command_data):
input_report = InputReport()
input_report.set_input_report_id(0x21)
joycontrol/report.py
+23 -3
View File
@@ -112,9 +112,11 @@ class InputReport:
for i in range(14, 50):
self.data[i] = 0x00
def set_mcu(self, data):
def set_ir_nfc_data(self, data):
if 50 + len(data) > len(self.data):
raise ValueError('Too much data.')
# write to data
data = bytes(data)
for i in range(len(data)):
self.data[50 + i] = data[i]
@@ -205,6 +207,15 @@ class InputReport:
else:
return bytes(self.data[:51])
def __str__(self):
_id = f'Input {self.get_input_report_id():x}'
_info = ''
if self.get_input_report_id() == 0x21:
_info = self.get_reply_to_subcommand_id()
_bytes = ' '.join(f'{byte:x}' for byte in bytes(self))
return f'{_id} {_info}\n{_bytes}'
class SubCommand(Enum):
REQUEST_DEVICE_INFO = 0x02
@@ -222,7 +233,7 @@ class SubCommand(Enum):
class OutputReportID(Enum):
SUB_COMMAND = 0x01
RUMBLE_ONLY = 0x10
REQUEST_MCU = 0x11
REQUEST_IR_NFC_MCU = 0x11
class OutputReport:
@@ -306,3 +317,12 @@ class OutputReport:
def __bytes__(self):
return bytes(self.data)
def __str__(self):
_id = f'Output {self.get_output_report_id()}'
_info = ''
if self.get_output_report_id() == OutputReportID.SUB_COMMAND:
_info = self.get_sub_command()
_bytes = ' '.join(f'{byte:x}' for byte in bytes(self))
return f'{_id} {_info}\n{_bytes}'
scripts/relay_joycon.py
+1
View File
@@ -20,6 +20,7 @@ PRODUCT_ID_JL = 8198
PRODUCT_ID_JR = 8199
PRODUCT_ID_PC = 8201
class Relay:
def __init__(self, capture_file=None):
self._capture_file = capture_file