relay/dev.py

#!/usr/bin/env python3
import platform
import shutil
import subprocess
import sys
import time

from datetime import datetime, timedelta
from importlib.util import find_spec
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Any, Sequence

try:
	import tomllib

except ImportError:
	if find_spec("toml") is None:
		subprocess.run([sys.executable, "-m", "pip", "install", "toml"])

	import toml as tomllib # type: ignore[no-redef]

if None in [find_spec("click"), find_spec("watchdog")]:
	CMD = [sys.executable, "-m", "pip", "install", "click >= 8.1.0", "watchdog >= 4.0.0"]
	PROC = subprocess.run(CMD, check = False)

	if PROC.returncode != 0:
		sys.exit()

	print("Successfully installed dependencies")

import click

from watchdog.observers import Observer
from watchdog.events import FileSystemEvent, PatternMatchingEventHandler


REPO = Path(__file__).parent
IGNORE_EXT = {
	'.py',
	'.pyc'
}


@click.group('cli')
def cli() -> None:
	'Useful commands for development'


@cli.command('install')
@click.option('--no-dev', '-d', is_flag = True, help = 'Do not install development dependencies')
def cli_install(no_dev: bool) -> None:
	with open('pyproject.toml', 'r', encoding = 'utf-8') as fd:
		data = tomllib.loads(fd.read())

	deps = data['project']['dependencies']
	deps.extend(data['project']['optional-dependencies']['dev'])

	subprocess.run([sys.executable, '-m', 'pip', 'install', '-U', *deps], check = False)


@cli.command('lint')
@click.argument('path', required = False, type = Path, default = REPO.joinpath('relay'))
@click.option('--watch', '-w', is_flag = True,
	help = 'Automatically, re-run the linters on source change')
def cli_lint(path: Path, watch: bool) -> None:
	path = path.expanduser().resolve()

	if watch:
		handle_run_watcher([sys.executable, "dev.py", "lint", str(path)], wait = True)
		return

	flake8 = [sys.executable, '-m', 'flake8', "dev.py", str(path)]
	mypy = [sys.executable, '-m', 'mypy', '--python-version', '3.12', 'dev.py', str(path)]

	click.echo('----- flake8 -----')
	subprocess.run(flake8)

	click.echo('\n\n----- mypy -----')
	subprocess.run(mypy)


@cli.command('clean')
def cli_clean() -> None:
	dirs = {
		'dist',
		'build',
		'dist-pypi'
	}

	for directory in dirs:
		shutil.rmtree(directory, ignore_errors = True)

	for path in REPO.glob('*.egg-info'):
		shutil.rmtree(path)

	for path in REPO.glob('*.spec'):
		path.unlink()


@cli.command('build')
def cli_build() -> None:
	from relay import __version__

	with TemporaryDirectory() as tmp:
		arch = 'amd64' if sys.maxsize >= 2**32 else 'i386'
		cmd = [
			sys.executable, '-m', 'PyInstaller',
			'--collect-data', 'relay',
			'--collect-data', 'aiohttp_swagger',
			'--hidden-import', 'pg8000',
			'--hidden-import', 'sqlite3',
			'--name', f'activityrelay-{__version__}-{platform.system().lower()}-{arch}',
			'--workpath', tmp,
			'--onefile', 'relay/__main__.py',
		]

		if platform.system() == 'Windows':
			cmd.append('--console')

			# putting the spec path on a different drive than the source dir breaks
			if str(REPO)[0] == tmp[0]:
				cmd.extend(['--specpath', tmp])

		else:
			cmd.append('--strip')
			cmd.extend(['--specpath', tmp])

		subprocess.run(cmd, check = False)


@cli.command('run')
@click.option('--dev', '-d', is_flag = True)
def cli_run(dev: bool) -> None:
	print('Starting process watcher')

	cmd = [sys.executable, '-m', 'relay', 'run']

	if dev:
		cmd.append('-d')

	handle_run_watcher(cmd, watch_path = REPO.joinpath("relay"))


def handle_run_watcher(
					*commands: Sequence[str],
					watch_path: Path | str = REPO,
					wait: bool = False) -> None:

	handler = WatchHandler(*commands, wait = wait)
	handler.run_procs()

	watcher = Observer()
	watcher.schedule(handler, str(watch_path), recursive=True) # type: ignore
	watcher.start() # type: ignore

	try:
		while True:
			time.sleep(1)

	except KeyboardInterrupt:
		pass

	handler.kill_procs()
	watcher.stop() # type: ignore
	watcher.join()


class WatchHandler(PatternMatchingEventHandler):
	patterns = ['*.py']


	def __init__(self, *commands: Sequence[str], wait: bool = False) -> None:
		PatternMatchingEventHandler.__init__(self) # type: ignore

		self.commands: Sequence[Sequence[str]] = commands
		self.wait: bool = wait
		self.procs: list[subprocess.Popen[Any]] = []
		self.last_restart: datetime = datetime.now()


	def kill_procs(self) -> None:
		for proc in self.procs:
			if proc.poll() is not None:
				continue

			print(f'Terminating process {proc.pid}')
			proc.terminate()
			sec = 0.0

			while proc.poll() is None:
				time.sleep(0.1)
				sec += 0.1

				if sec >= 5:
					print('Failed to terminate. Killing process...')
					proc.kill()
					break

			print('Process terminated')


	def run_procs(self, restart: bool = False) -> None:
		if restart:
			if datetime.now() - timedelta(seconds = 3) < self.last_restart:
				return

			self.kill_procs()

		self.last_restart = datetime.now()

		if self.wait:
			self.procs = []

			for cmd in self.commands:
				print('Running command:', ' '.join(cmd))
				subprocess.run(cmd)

		else:
			self.procs = list(subprocess.Popen(cmd) for cmd in self.commands)
			pids = (str(proc.pid) for proc in self.procs)
			print('Started processes with PIDs:', ', '.join(pids))


	def on_any_event(self, event: FileSystemEvent) -> None:
		if event.event_type not in ['modified', 'created', 'deleted']:
			return

		self.run_procs(restart = True)


if __name__ == '__main__':
	cli()