saysynth.core.controller

Utilities for registering processes as files under ~/.saysynth/pid so they can be dynamically stopped.

  1"""
  2Utilities for registering processes as files under `~/.saysynth/pid` so they can be dynamically stopped.
  3<center><img src="/assets/img/spiral.png"></img></center>
  4"""
  5import os
  6import signal
  7from pathlib import Path
  8from typing import Dict, List, Optional, Union
  9
 10import click
 11
 12from saysynth.cli.colors import blue, green, yellow
 13from saysynth.constants import DEFAULT_SEQUENCE_NAME
 14from saysynth.utils import random_track_name
 15
 16SEQUENCE_PID_LOG = os.path.expanduser("~/.saysynth/pid")
 17"""
 18Where to log child pids of parent processes.
 19These will take the form of `~/.saysynth/pid/{seq}.{track}.{audio_device}.{parent_pid}`
 20"""
 21
 22
 23def _read_pid_file(path: str) -> List[int]:
 24    with open(path, "r") as f:
 25        return list([int(line.strip()) for line in f.readlines()])
 26
 27
 28def _append_pid_file(path: str, pid: int) -> None:
 29    with open(path, "a") as f:
 30        f.write(str(pid) + "\n")
 31
 32
 33def _write_pid_file(path: str, pids: List[int]) -> None:
 34    with open(path, "w") as f:
 35        f.write("\n".join([str(p) for p in pids]))
 36
 37
 38def ensure_pid_log() -> None:
 39    if not os.path.exists(SEQUENCE_PID_LOG):
 40        os.makedirs(SEQUENCE_PID_LOG)
 41
 42
 43def _list_pid_file_paths(
 44    seq: Optional[str] = None,
 45    track: Optional[str] = None,
 46    ad: Optional[str] = None,
 47    parent_pid: Optional[int] = None,
 48) -> List[Path]:
 49    """
 50    List pid file paths by seq, track, and/or pid.
 51
 52    Args:
 53        seq: An optional sequence name to filter file paths by
 54        track: An optional track name to filter file paths by
 55        ad: An optional audio device to filter file paths by
 56        parent_pid: An optional parent_pid to filter file paths by
 57    """
 58    return Path(SEQUENCE_PID_LOG).glob(
 59        f'{seq or "*"}.{track or "*"}.{ad or "*"}.{parent_pid or "*"}'
 60    )
 61
 62
 63def list_pids() -> List[Dict[str, Union[str, int, List[int]]]]:
 64    """
 65    List and parse all pid file paths and lookup the child pids.
 66    """
 67    pids = []
 68    for path in _list_pid_file_paths():
 69        seq, track, ad, parent_pid = str(path).split("/")[-1].split(".")
 70        child_pids = lookup_child_pids(seq, track, ad, parent_pid)
 71        pids.append(
 72            {
 73                "seq": seq if seq != DEFAULT_SEQUENCE_NAME else "none",
 74                "track": track,
 75                "ad": ad if ad != "None" else "default",
 76                "parent_pid": parent_pid,
 77                "child_pids": child_pids,
 78            }
 79        )
 80    return sorted(pids, key=lambda x: x["seq"] + x["track"])
 81
 82
 83def add_parent_pid(seq: str, track: str, ad: str, parent_pid: int) -> None:
 84    """
 85    Associate a pid with a track wihin a sequence.
 86
 87    Args:
 88        seq: An optional sequence name to associate with the parent pid
 89        track: An optional track name to associate with the parent pid
 90        ad: An optional audio device to associate with the parent pid
 91        parent_pid: The parent pid
 92    """
 93    ensure_pid_log()
 94    path = f"{SEQUENCE_PID_LOG}/{seq}.{track}.{ad}.{parent_pid}"
 95    Path(path).touch()
 96    return path
 97
 98
 99def rm_parent_pid(
100    seq: Optional[str] = None,
101    track: Optional[str] = None,
102    ad: Optional[str] = None,
103    parent_pid: Optional[int] = None,
104) -> None:
105    """
106    Remove pid log files for a seq and/or track
107
108    Args:
109        seq: An optional sequence name to remove pids by
110        track: An optional track name to remove pids by
111        ad: An optional audio device to remove pids by
112        parent_pid: An optional parent_pid to remove pids by
113    """
114    for path in _list_pid_file_paths(seq, track, ad, parent_pid):
115        path.unlink()
116
117
118def add_child_pid(
119    child_pid: int, parent_pid: int, parent_pid_file: Optional[str]
120) -> None:
121    """
122    Add a child pid to a parent_pid.
123
124    Args:
125        child_pid: The child process to register with the parent.
126        parent_pid: The parent pid.
127    """
128    if parent_pid_file:
129        paths = [parent_pid_file]
130    else:
131        paths = _list_pid_file_paths(parent_pid=parent_pid)
132    for path in paths:
133        _append_pid_file(path, child_pid)
134
135
136def rm_child_pid(child_pid: int, parent_pid: int) -> None:
137    """
138    Remove a child pid from a parent_pid.
139
140    Args:
141        child_pid: The child process to dergisister from the parent.
142        parent_pid: The parent pid.
143    """
144    paths = _list_pid_file_paths(parent_pid=parent_pid)
145    for path in paths:
146        pids = set(_read_pid_file(path))
147        pids.remove(child_pid)
148        _write_pid_file(path, pids)
149
150
151def lookup_parent_pids(
152    seq: Optional[str] = None,
153    track: Optional[str] = None,
154    ad: Optional[str] = None,
155    parent_pid: Optional[int] = None,
156) -> List[int]:
157    """
158    Lookup all of the parent pids for a seq and/or track.
159
160    Args:
161        seq: An optional sequence name to filter parent pids by
162        track: An optional track name to filter parent pids by
163        ad: An optional audio device to filter parent pids by
164        parent_pid: An optional parent_pid to filter parent pids by
165    """
166    return [
167        int(str(path).split(".")[-1])
168        for path in _list_pid_file_paths(seq, track, ad, parent_pid)
169    ]
170
171
172def lookup_child_pids(
173    seq: Optional[str] = None,
174    track: Optional[str] = None,
175    ad: Optional[str] = None,
176    parent_pid: Optional[int] = None,
177) -> List[int]:
178    """
179    Lookup the child pids for a seq and/or track.
180
181    Args:
182        seq: An optional sequence name to filter child pids by
183        track: An optional track name to filter child pids by
184        ad: An optional audio device to filter child pids by
185        parent_pid: An optional parent_pid to filter child pids by
186    """
187    pids = []
188    for path in _list_pid_file_paths(seq, track, ad, parent_pid):
189        pids.extend(_read_pid_file(path))
190    return list(set(pids))
191
192
193def lookup_pids(
194    seq: Optional[str] = None,
195    track: Optional[str] = None,
196    ad: Optional[str] = None,
197    parent_pid: Optional[int] = None,
198) -> None:
199    """
200    Lookup all pids for a sequence / track.
201
202    Args:
203        seq: An optional sequence name to filter pids by
204        track: An optional track name to filter pids by
205        ad: An optional audio device to filter pids by
206        parent_pid: An optional parent_pid to filter pids by
207    """
208    return lookup_parent_pids(seq, track, ad, parent_pid) + lookup_child_pids(
209        seq, track, ad, parent_pid
210    )
211
212
213def stop_child_pids(
214    seq: Optional[str] = None,
215    track: Optional[str] = None,
216    ad: Optional[str] = None,
217    parent_pid: Optional[int] = None,
218) -> None:
219    """
220    Stop all the child pids of a parent.
221
222    Args:
223        seq: An optional sequence name to stop child pids by
224        track: An optional track name to stop child pids by
225        ad: An optional audio device to stop child pids by
226        parent_pid: An optional parent_pid to stop child pids by
227    """
228    pids = lookup_child_pids(seq, track, ad, parent_pid)
229    for pid in pids:
230        try:
231            os.kill(pid, signal.SIGTERM)
232        except ProcessLookupError:
233            pass
234    rm_parent_pid(seq, track, ad, parent_pid)
235
236
237def handle_cli_options(command, **kwargs) -> dict:
238    """
239    Initialize controller and set cli options.
240
241    Args:
242        command: The name of the cli command (eg: `chord`)
243    """
244    text = kwargs.get('text', None)
245    # check for text as filepath.
246    if text and os.path.exists(os.path.expanduser(text)):
247        with open(text, 'r') as f:
248            kwargs['text'] = f.read().strip()
249    parent_pid = os.getpid()
250    track_name = random_track_name(command, **kwargs)
251    add_parent_pid(track_name, command, kwargs.get("audio_device"), parent_pid)
252    click.echo(
253        f"▶️ {green('starting')} {blue(track_name)} with {yellow('pid')}: {blue(str(parent_pid))}",
254        err=True,
255    )
256    kwargs["parent_pid"] = parent_pid
257    return kwargs
SEQUENCE_PID_LOG = '/root/.saysynth/pid'

Where to log child pids of parent processes. These will take the form of ~/.saysynth/pid/{seq}.{track}.{audio_device}.{parent_pid}

def ensure_pid_log() -> None:
39def ensure_pid_log() -> None:
40    if not os.path.exists(SEQUENCE_PID_LOG):
41        os.makedirs(SEQUENCE_PID_LOG)
def list_pids() -> List[Dict[str, Union[str, int, List[int]]]]:
64def list_pids() -> List[Dict[str, Union[str, int, List[int]]]]:
65    """
66    List and parse all pid file paths and lookup the child pids.
67    """
68    pids = []
69    for path in _list_pid_file_paths():
70        seq, track, ad, parent_pid = str(path).split("/")[-1].split(".")
71        child_pids = lookup_child_pids(seq, track, ad, parent_pid)
72        pids.append(
73            {
74                "seq": seq if seq != DEFAULT_SEQUENCE_NAME else "none",
75                "track": track,
76                "ad": ad if ad != "None" else "default",
77                "parent_pid": parent_pid,
78                "child_pids": child_pids,
79            }
80        )
81    return sorted(pids, key=lambda x: x["seq"] + x["track"])

List and parse all pid file paths and lookup the child pids.

def add_parent_pid(seq: str, track: str, ad: str, parent_pid: int) -> None:
84def add_parent_pid(seq: str, track: str, ad: str, parent_pid: int) -> None:
85    """
86    Associate a pid with a track wihin a sequence.
87
88    Args:
89        seq: An optional sequence name to associate with the parent pid
90        track: An optional track name to associate with the parent pid
91        ad: An optional audio device to associate with the parent pid
92        parent_pid: The parent pid
93    """
94    ensure_pid_log()
95    path = f"{SEQUENCE_PID_LOG}/{seq}.{track}.{ad}.{parent_pid}"
96    Path(path).touch()
97    return path

Associate a pid with a track wihin a sequence.

Arguments:
  • seq: An optional sequence name to associate with the parent pid
  • track: An optional track name to associate with the parent pid
  • ad: An optional audio device to associate with the parent pid
  • parent_pid: The parent pid
def rm_parent_pid( seq: Optional[str] = None, track: Optional[str] = None, ad: Optional[str] = None, parent_pid: Optional[int] = None) -> None:
100def rm_parent_pid(
101    seq: Optional[str] = None,
102    track: Optional[str] = None,
103    ad: Optional[str] = None,
104    parent_pid: Optional[int] = None,
105) -> None:
106    """
107    Remove pid log files for a seq and/or track
108
109    Args:
110        seq: An optional sequence name to remove pids by
111        track: An optional track name to remove pids by
112        ad: An optional audio device to remove pids by
113        parent_pid: An optional parent_pid to remove pids by
114    """
115    for path in _list_pid_file_paths(seq, track, ad, parent_pid):
116        path.unlink()

Remove pid log files for a seq and/or track

Arguments:
  • seq: An optional sequence name to remove pids by
  • track: An optional track name to remove pids by
  • ad: An optional audio device to remove pids by
  • parent_pid: An optional parent_pid to remove pids by
def add_child_pid(child_pid: int, parent_pid: int, parent_pid_file: Optional[str]) -> None:
119def add_child_pid(
120    child_pid: int, parent_pid: int, parent_pid_file: Optional[str]
121) -> None:
122    """
123    Add a child pid to a parent_pid.
124
125    Args:
126        child_pid: The child process to register with the parent.
127        parent_pid: The parent pid.
128    """
129    if parent_pid_file:
130        paths = [parent_pid_file]
131    else:
132        paths = _list_pid_file_paths(parent_pid=parent_pid)
133    for path in paths:
134        _append_pid_file(path, child_pid)

Add a child pid to a parent_pid.

Arguments:
  • child_pid: The child process to register with the parent.
  • parent_pid: The parent pid.
def rm_child_pid(child_pid: int, parent_pid: int) -> None:
137def rm_child_pid(child_pid: int, parent_pid: int) -> None:
138    """
139    Remove a child pid from a parent_pid.
140
141    Args:
142        child_pid: The child process to dergisister from the parent.
143        parent_pid: The parent pid.
144    """
145    paths = _list_pid_file_paths(parent_pid=parent_pid)
146    for path in paths:
147        pids = set(_read_pid_file(path))
148        pids.remove(child_pid)
149        _write_pid_file(path, pids)

Remove a child pid from a parent_pid.

Arguments:
  • child_pid: The child process to dergisister from the parent.
  • parent_pid: The parent pid.
def lookup_parent_pids( seq: Optional[str] = None, track: Optional[str] = None, ad: Optional[str] = None, parent_pid: Optional[int] = None) -> List[int]:
152def lookup_parent_pids(
153    seq: Optional[str] = None,
154    track: Optional[str] = None,
155    ad: Optional[str] = None,
156    parent_pid: Optional[int] = None,
157) -> List[int]:
158    """
159    Lookup all of the parent pids for a seq and/or track.
160
161    Args:
162        seq: An optional sequence name to filter parent pids by
163        track: An optional track name to filter parent pids by
164        ad: An optional audio device to filter parent pids by
165        parent_pid: An optional parent_pid to filter parent pids by
166    """
167    return [
168        int(str(path).split(".")[-1])
169        for path in _list_pid_file_paths(seq, track, ad, parent_pid)
170    ]

Lookup all of the parent pids for a seq and/or track.

Arguments:
  • seq: An optional sequence name to filter parent pids by
  • track: An optional track name to filter parent pids by
  • ad: An optional audio device to filter parent pids by
  • parent_pid: An optional parent_pid to filter parent pids by
def lookup_child_pids( seq: Optional[str] = None, track: Optional[str] = None, ad: Optional[str] = None, parent_pid: Optional[int] = None) -> List[int]:
173def lookup_child_pids(
174    seq: Optional[str] = None,
175    track: Optional[str] = None,
176    ad: Optional[str] = None,
177    parent_pid: Optional[int] = None,
178) -> List[int]:
179    """
180    Lookup the child pids for a seq and/or track.
181
182    Args:
183        seq: An optional sequence name to filter child pids by
184        track: An optional track name to filter child pids by
185        ad: An optional audio device to filter child pids by
186        parent_pid: An optional parent_pid to filter child pids by
187    """
188    pids = []
189    for path in _list_pid_file_paths(seq, track, ad, parent_pid):
190        pids.extend(_read_pid_file(path))
191    return list(set(pids))

Lookup the child pids for a seq and/or track.

Arguments:
  • seq: An optional sequence name to filter child pids by
  • track: An optional track name to filter child pids by
  • ad: An optional audio device to filter child pids by
  • parent_pid: An optional parent_pid to filter child pids by
def lookup_pids( seq: Optional[str] = None, track: Optional[str] = None, ad: Optional[str] = None, parent_pid: Optional[int] = None) -> None:
194def lookup_pids(
195    seq: Optional[str] = None,
196    track: Optional[str] = None,
197    ad: Optional[str] = None,
198    parent_pid: Optional[int] = None,
199) -> None:
200    """
201    Lookup all pids for a sequence / track.
202
203    Args:
204        seq: An optional sequence name to filter pids by
205        track: An optional track name to filter pids by
206        ad: An optional audio device to filter pids by
207        parent_pid: An optional parent_pid to filter pids by
208    """
209    return lookup_parent_pids(seq, track, ad, parent_pid) + lookup_child_pids(
210        seq, track, ad, parent_pid
211    )

Lookup all pids for a sequence / track.

Arguments:
  • seq: An optional sequence name to filter pids by
  • track: An optional track name to filter pids by
  • ad: An optional audio device to filter pids by
  • parent_pid: An optional parent_pid to filter pids by
def stop_child_pids( seq: Optional[str] = None, track: Optional[str] = None, ad: Optional[str] = None, parent_pid: Optional[int] = None) -> None:
214def stop_child_pids(
215    seq: Optional[str] = None,
216    track: Optional[str] = None,
217    ad: Optional[str] = None,
218    parent_pid: Optional[int] = None,
219) -> None:
220    """
221    Stop all the child pids of a parent.
222
223    Args:
224        seq: An optional sequence name to stop child pids by
225        track: An optional track name to stop child pids by
226        ad: An optional audio device to stop child pids by
227        parent_pid: An optional parent_pid to stop child pids by
228    """
229    pids = lookup_child_pids(seq, track, ad, parent_pid)
230    for pid in pids:
231        try:
232            os.kill(pid, signal.SIGTERM)
233        except ProcessLookupError:
234            pass
235    rm_parent_pid(seq, track, ad, parent_pid)

Stop all the child pids of a parent.

Arguments:
  • seq: An optional sequence name to stop child pids by
  • track: An optional track name to stop child pids by
  • ad: An optional audio device to stop child pids by
  • parent_pid: An optional parent_pid to stop child pids by
def handle_cli_options(command, **kwargs) -> dict:
238def handle_cli_options(command, **kwargs) -> dict:
239    """
240    Initialize controller and set cli options.
241
242    Args:
243        command: The name of the cli command (eg: `chord`)
244    """
245    text = kwargs.get('text', None)
246    # check for text as filepath.
247    if text and os.path.exists(os.path.expanduser(text)):
248        with open(text, 'r') as f:
249            kwargs['text'] = f.read().strip()
250    parent_pid = os.getpid()
251    track_name = random_track_name(command, **kwargs)
252    add_parent_pid(track_name, command, kwargs.get("audio_device"), parent_pid)
253    click.echo(
254        f"▶️ {green('starting')} {blue(track_name)} with {yellow('pid')}: {blue(str(parent_pid))}",
255        err=True,
256    )
257    kwargs["parent_pid"] = parent_pid
258    return kwargs

Initialize controller and set cli options.

Arguments:
  • command: The name of the cli command (eg: chord)