saysynth.core.controller

  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