add cli parameter for default actuator power (#29)

Co-authored-by: Mæve Rey <42996147M-rey@users.noreply.github.com>
This commit is contained in:
Nova 2023-03-14 00:48:53 +01:00 committed by GitHub
parent 72f424e9b8
commit 09045cdc82
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 38 additions and 9 deletions

View file

@ -11,7 +11,7 @@ from fediplug.cli import options
from typing import Tuple from typing import Tuple
async def connect_plug_client(websocket_url: str = "127.0.0.1:12345") -> Client: async def connect_plug_client(websocket_url: str) -> Client:
"""create Client object and connect plug client to Intiface Central or similar""" """create Client object and connect plug client to Intiface Central or similar"""
plug_client = Client("fediplug", ProtocolSpec.v3) plug_client = Client("fediplug", ProtocolSpec.v3)
connector = WebsocketConnector(f"ws://{websocket_url}", logger=plug_client.logger) connector = WebsocketConnector(f"ws://{websocket_url}", logger=plug_client.logger)

View file

@ -88,10 +88,23 @@ def login(instance: str) -> None:
@cli.command() @cli.command()
@click.option("-w", "--websocket", help="Pass an alternative websocket.") @click.option(
"-p",
"--actuator_power",
help="sets the default actuator power, if no value is given.",
default="100%",
metavar=r"0%-100%",
)
@click.option(
"-w",
"--websocket",
help="Pass an alternative websocket.",
default="127.0.0.1:12345",
metavar="ip:port",
)
@click.argument("instance") @click.argument("instance")
@click.argument("users", nargs=-1) @click.argument("users", nargs=-1)
def stream(instance: str, users: Tuple[str], websocket: str): def stream(instance: str, users: Tuple[str], actuator_power: str, websocket: str):
"""Control buttplug.io device from your timeline.""" """Control buttplug.io device from your timeline."""
event_loop = asyncio.get_event_loop() event_loop = asyncio.get_event_loop()
@ -104,5 +117,12 @@ def stream(instance: str, users: Tuple[str], websocket: str):
access_token = get_access_token(instance) access_token = get_access_token(instance)
mastodon.stream( mastodon.stream(
instance, users, client_id, client_secret, access_token, plug_client, event_loop instance,
users,
client_id,
client_secret,
access_token,
plug_client,
event_loop,
actuator_power,
) )

View file

@ -31,11 +31,13 @@ class StreamListener(mastodon.StreamListener):
instance: str, instance: str,
users: list[str], users: list[str],
event_loop: object, event_loop: object,
actuator_power: str,
) -> None: ) -> None:
self.plug_client = plug_client self.plug_client = plug_client
self.instance = instance self.instance = instance
self.users = users self.users = users
self.event_loop = event_loop self.event_loop = event_loop
self.actuator_power = actuator_power
self.regular_expression = re.compile(r"((?:\b(?:\d+s)(?:\s|\b))+(?:\d+%)?)+") self.regular_expression = re.compile(r"((?:\b(?:\d+s)(?:\s|\b))+(?:\d+%)?)+")
# extracts commands from captured toots for usage in buttplug.io actuator # extracts commands from captured toots for usage in buttplug.io actuator
# if no power % is given, a default will be set later # if no power % is given, a default will be set later
@ -79,7 +81,7 @@ class StreamListener(mastodon.StreamListener):
"""Here we extract the instructions for the butplug""" """Here we extract the instructions for the butplug"""
buttplug_instructions = extract_buttplug_instructions( buttplug_instructions = extract_buttplug_instructions(
status, self.regular_expression status, self.regular_expression, self.actuator_power
) )
if buttplug_instructions: # check if buttplug_instructions is not empty if buttplug_instructions: # check if buttplug_instructions is not empty
for buttplug_instruction in buttplug_instructions: for buttplug_instruction in buttplug_instructions:
@ -127,13 +129,20 @@ def login(instance, client_id, client_secret, grant_code):
def stream( def stream(
instance, users, client_id, client_secret, access_token, plug_client, event_loop instance,
users,
client_id,
client_secret,
access_token,
plug_client,
event_loop,
actuator_power,
): ):
"""Stream statuses and add them to a queue.""" """Stream statuses and add them to a queue."""
client = build_client(instance, client_id, client_secret, access_token) client = build_client(instance, client_id, client_secret, access_token)
users = [normalize_username(user, instance) for user in users] users = [normalize_username(user, instance) for user in users]
listener = StreamListener(plug_client, instance, users, event_loop) listener = StreamListener(plug_client, instance, users, event_loop, actuator_power)
click.echo(f"==> Streaming from {instance}") click.echo(f"==> Streaming from {instance}")
client.stream_user(listener) client.stream_user(listener)
@ -157,7 +166,7 @@ def normalize_username(user, instance):
return user return user
def extract_buttplug_instructions(status, regular_expression): def extract_buttplug_instructions(status, regular_expression, actuator_power):
"""Extract buttplug instruction informations from a toot.""" """Extract buttplug instruction informations from a toot."""
toot = lh.fromstring(status["content"]) toot = lh.fromstring(status["content"])
toot = clean_html(toot) toot = clean_html(toot)
@ -170,7 +179,7 @@ def extract_buttplug_instructions(status, regular_expression):
commands = instruction.strip().split(" ") commands = instruction.strip().split(" ")
print(commands) print(commands)
if commands[-1][-1] != "%": if commands[-1][-1] != "%":
commands.append("100%") commands.append(actuator_power)
commands = [int(command[:-1]) for command in commands] commands = [int(command[:-1]) for command in commands]
power = commands.pop() / 100 # convert power from % to range 0..1 power = commands.pop() / 100 # convert power from % to range 0..1
commands = list(zip(commands, cycle([power]))) commands = list(zip(commands, cycle([power])))