New Session CLI Command

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "1.1.0"
+current_version = "1.2.0"
 commit = true
 message = "Release v{new_version}"
 tag = true

.gitignore

@@ -3,3 +3,4 @@ __pycache__/
 /.tox
 /docs/_build/
 /dist
+/.venv

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## v1.2.0 (2025-01-14)
+- Add session mode to CLI. Control a session declaratively with a JSON file as input
+- Updated docs for session mode usage
+
 ## v1.1.0 (2024-08-17)
 
 - New random mode features:

docs/cli.md

@@ -159,6 +159,81 @@ can be used to customize the behavior (default values are shown):
 - `--spam-intensity`: Intensity for spam shocks (by default, the given
   `--intensity` is used)
 
+## Session mode
+
+To schedule shock, spam, vibrate and beep events over a session, the session
+command takes a json file as input allowing automation of multiple shockers for
+completely handsfree use.
+
+```console
+$ pishock session examples/session.json
+✔ Validating events
+✔ Event list is valid
+▶ Session started
+🟩  Count in mode is set to beep
+🔔 Beeping shocker1 for 1s
+🔔 Beeping shocker1 for 1s
+🔔 Beeping shocker1 for 1s
+🕐 Max runtime is 3600 seconds
+🕐 Spam Cooldown is 120 seconds
+⚡ Shocking shocker1 for 9s at 82%
+⚡ Shocking shocker2 for 14s at 55%
+```
+
+Following similar field names to the random mode, the JSON format is as follows:
+
+```js
+{
+  "shocker_names": ["shocker-1", "shocker-2"], // cli shocker names
+  "max_runtime": "1h", // automatically end the session (default 1h inclusive of init_delay)
+  "init_delay": "2m", // delay the start of the script
+  "spam_cooldown": "2m", // limit how much you can be spammed
+  "count_in_mode": "beep", // if specified, the script will count down to session start
+  "events": [ // a list of events. add as many breakpoints as needed
+    {
+      "time": "0", // a time in seconds for when the session changes
+      "sync_mode": "sync", // random-shocker, sync, round-robin, dealers-choice
+      "break_duration": "1-10", // add spaces between shocker operations
+      "vibrate": { // by default, the program will vibe randomly between events
+        "intensity": "20-60", // intensity out of 100
+        "duration": "1-6" // duration in seconds (1-15)
+      },
+      "shock": {
+        "possibility": 15, // the percent chance of this event happening
+        "intensity": "3-8",
+        "duration": "1-6"
+      },
+      "spam": {
+        "possibility": 1,
+        "operations": "10-20", // how many times to send the shocks consecutively
+        "intensity": "3-5",
+        "duration": "1-2",
+        "delay": 0.3 // delay in seconds between spammed shocks
+      },
+      "beep": {
+        "possibility": 5,
+        "duration": "1-6"
+      }
+    },
+    {
+      "time": "60s", // the session will change after the specified time
+      "sync_mode": "random-shocker", // change the way your shockers are chosen with each step
+      "vibrate": {
+        "possibility": 5,
+        "intensity": "70-90",
+        "duration": "3-6"
+      }
+      "shock": {
+        "possibility": 5,
+        "intensity": "1-2",
+        "duration": "12-15"
+      }
+  },
+  ... // define as many events as needed
+  ]
+}
+```
+
 ## Serial usage
 
 When a PiShock is attached via USB, the commands `shock`, `vibrate`, `beep` and

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pishock"
-version = "1.1.0"
+version = "1.2.0"
 description = "The bestest Python PiShock API wrapper and CLI!"
 authors = ["Zerario <[email protected]>"]
 readme = "README.md"

src/pishock/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "1.1.0"
+__version__ = "1.2.0"
 
 from .zap.core import (
     Shocker as Shocker,

src/pishock/zap/cli/cli.py

@@ -1,5 +1,6 @@
 import contextlib
 import difflib
+import json
 import pathlib
 import random
 import sys
@@ -15,7 +16,7 @@
 from typing_extensions import Annotated, TypeAlias
 
 from pishock.zap import serialapi, core, httpapi
-from pishock.zap.cli import cli_random, cli_serial, cli_utils, cli_code
+from pishock.zap.cli import cli_random, cli_serial, cli_utils, cli_code, cli_session
 
 """Command-line interface for PiShock."""
 
@@ -55,6 +56,12 @@
         metavar="ID",
     ),
 ]
+SessionFileArg: TypeAlias = Annotated[
+    str,
+    typer.Argument(
+        help="A path to a JSON file containing the session format"
+    )
+]
 DurationOpt: TypeAlias = Annotated[
     float,
     typer.Option("-d", "--duration", min=0, help="Duration in seconds."),
@@ -66,7 +73,6 @@
     ),
 ]
 
-
 @contextlib.contextmanager
 def handle_errors(*args: Type[Exception]) -> Iterator[None]:
     try:
@@ -300,6 +306,26 @@ def random_mode(
     )
     random_shocker.run()
 
+@app.command(name="session")
+def session_mode(
+    ctx: typer.Context,
+    json_file: SessionFileArg) -> None: 
+    """Use a session workflow to have haptic/shock/beep operations change over time"""
+    if not json_file:
+        raise typer.BadParameter("Must provide a session file path in JSON format")
+
+    with open(json_file, 'r', encoding="UTF-8") as file:
+        data = json.load(file)
+        shocker_objs = [get_shocker(ctx.obj, shocker) for shocker in data.get("shocker_names")]
+
+        session = cli_session.Session(
+            shockers=shocker_objs,
+            data=data
+        )
+
+        session.validate_events()
+
+        session.run()
 
 @app.command(rich_help_panel="API credentials")
 def verify(ctx: typer.Context) -> None:

src/pishock/zap/cli/cli_random.py

@@ -1,92 +1,14 @@
 import contextlib
 import dataclasses
 import random
-import re
 import time
-from typing import Iterator, List, Optional, Union, Callable
-
-import click
+from typing import Iterator, List, Optional
 import rich
 import typer
 from typing_extensions import Annotated, TypeAlias
-
 from pishock.zap import httpapi, core
 from pishock.zap.cli import cli_utils as utils
 
-
-class RangeParser(click.ParamType):
-    name = "Range"
-
-    def __init__(
-        self, min: int, max: Optional[int] = None, converter: Callable[[str], int] = int
-    ) -> None:
-        self.min = min
-        self.max = max
-        self.converter = converter
-
-    def _parse_single(self, s: str) -> int:
-        try:
-            n = self.converter(s)
-        except ValueError:
-            self.fail(f"Value must be a {self.converter.__name__}: {s}")
-
-        if self.max is None and n < self.min:
-            self.fail(f"Value must be at least {self.min}: {n}")
-        if self.max is not None and not (self.min <= n <= self.max):
-            self.fail(f"Value must be between {self.min} and {self.max}: {n}")
-
-        return n
-
-    def convert(
-        self,
-        value: Union[str, utils.Range],
-        param: Optional[click.Parameter],
-        ctx: Optional[click.Context],
-    ) -> utils.Range:
-        if isinstance(value, utils.Range):  # default value
-            return value
-
-        if "-" not in value:
-            n = self._parse_single(value)
-            return utils.Range(n, n)
-
-        if value.count("-") > 1:
-            self.fail("Range must be in the form min-max.")
-
-        a_str, b_str = value.split("-")
-        a = self._parse_single(a_str)
-        b = self._parse_single(b_str)
-
-        try:
-            return utils.Range(a, b)
-        except ValueError as e:
-            self.fail(str(e))
-
-
-def parse_duration(duration: str) -> int:
-    """Parse duration in format XhYmZs into second duration."""
-    if duration.isdigit():
-        return int(duration)
-
-    match = re.fullmatch(
-        r"(?P<hours>[0-9]+(\.[0-9])?h)?\s*"
-        r"(?P<minutes>[0-9]+(\.[0-9])?m)?\s*"
-        r"(?P<seconds>[0-9]+(\.[0-9])?s)?",
-        duration,
-    )
-    if not match or not match.group(0):
-        raise ValueError(
-            f"Invalid duration: {duration} - " "expected XhYmZs or a number of seconds"
-        )
-    seconds_string = match.group("seconds") if match.group("seconds") else "0"
-    seconds = float(seconds_string.rstrip("s"))
-    minutes_string = match.group("minutes") if match.group("minutes") else "0"
-    minutes = float(minutes_string.rstrip("m"))
-    hours_string = match.group("hours") if match.group("hours") else "0"
-    hours = float(hours_string.rstrip("h"))
-    return int(seconds + minutes * 60 + hours * 3600)
-
-
 @dataclasses.dataclass
 class SpamSettings:
     possibility: int
@@ -221,7 +143,7 @@ def _tick(self) -> None:
             "Duration in seconds, as a single value or a min-max range (0-15 "
             "respectively)."
         ),
-        click_type=RangeParser(min=0, max=15),
+        click_type=utils.RangeParser(min=0, max=15),
     ),
 ]
 
@@ -234,7 +156,7 @@ def _tick(self) -> None:
             "Intensity in percent, as a single value or min-max range (0-100 "
             "respectively)."
         ),
-        click_type=RangeParser(min=0, max=100),
+        click_type=utils.RangeParser(min=0, max=100),
     ),
 ]
 
@@ -246,7 +168,7 @@ def _tick(self) -> None:
         help="Delay between operations, in seconds or a string like "
         "1h2m3s (with h/m being optional). With a min-max range of such values, "
         "picked randomly.",
-        click_type=RangeParser(min=0, converter=parse_duration),
+        click_type=utils.RangeParser(min=0, converter=utils.parse_duration),
     ),
 ]
 
@@ -257,7 +179,7 @@ def _tick(self) -> None:
         help="Initial delay before the first operation, in seconds or a string like "
         "1h2m3s (with h/m being optional). With a min-max range of such values, "
         "picked randomly.",
-        click_type=RangeParser(min=0, converter=parse_duration),
+        click_type=utils.RangeParser(min=0, converter=utils.parse_duration),
     ),
 ]
 
@@ -274,7 +196,7 @@ def _tick(self) -> None:
     utils.Range,
     typer.Option(
         help="Number of operations to spam, as a single value or min-max range.",
-        click_type=RangeParser(min=1),
+        click_type=utils.RangeParser(min=1),
     ),
 ]
 
@@ -284,7 +206,7 @@ def _tick(self) -> None:
         help="Delay between spam operations, in seconds or a string like "
         "1h2m3s (with h/m being optional). With a min-max range of such values, "
         "picked randomly.",
-        click_type=RangeParser(min=0, converter=parse_duration),
+        click_type=utils.RangeParser(min=0, converter=utils.parse_duration),
     ),
 ]
 
@@ -295,7 +217,7 @@ def _tick(self) -> None:
             "Duration of spam operations in seconds, as a single value or min-max "
             "range."
         ),
-        click_type=RangeParser(min=0, max=15),
+        click_type=utils.RangeParser(min=0, max=15),
     ),
 ]
 
@@ -306,7 +228,7 @@ def _tick(self) -> None:
             "Intensity of spam operations in percent, as a single value or min-max "
             "range. If not given, normal intensity is used."
         ),
-        click_type=RangeParser(min=0, max=100),
+        click_type=utils.RangeParser(min=0, max=100),
     ),
 ]
 
@@ -317,7 +239,7 @@ def _tick(self) -> None:
             "Maximum runtime in seconds or a string like 1h2m3s (with h/m being "
             "optional). With a min-max range of such values, picked randomly."
         ),
-        click_type=RangeParser(min=0, converter=parse_duration),
+        click_type=utils.RangeParser(min=0, converter=utils.parse_duration),
     ),
 ]
 
@@ -328,7 +250,7 @@ def _tick(self) -> None:
             "Duration for vibration in seconds, as a single value or a min-max "
             "range (0-15 respectively). If not given, --duration is used."
         ),
-        click_type=RangeParser(min=0, max=15),
+        click_type=utils.RangeParser(min=0, max=15),
     ),
 ]
 
@@ -339,7 +261,7 @@ def _tick(self) -> None:
             "Intensity in percent, as a single value or min-max range (0-100 "
             "respectively). If not given, --intensity is used."
         ),
-        click_type=RangeParser(min=0, max=100),
+        click_type=utils.RangeParser(min=0, max=100),
     ),
 ]