more typing work
This commit is contained in:
Kovid Goyal
parent
b6692849d6
commit
9f2fb76309
@ -8,10 +8,20 @@ import pwd
|
||||
import sys
|
||||
from contextlib import suppress
|
||||
from functools import lru_cache
|
||||
from typing import Optional, Set, NamedTuple
|
||||
from typing import TYPE_CHECKING, NamedTuple, Optional, Set
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .options_stub import Options # noqa
|
||||
|
||||
|
||||
class Version(NamedTuple):
|
||||
major: int
|
||||
minor: int
|
||||
patch: int
|
||||
|
||||
|
||||
appname = 'kitty'
|
||||
version = (0, 16, 0)
|
||||
version = Version(0, 16, 0)
|
||||
str_version = '.'.join(map(str, version))
|
||||
_plat = sys.platform.lower()
|
||||
is_macos = 'darwin' in _plat
|
||||
@ -37,7 +47,7 @@ class WindowGeometry(NamedTuple):
|
||||
|
||||
|
||||
@lru_cache(maxsize=2)
|
||||
def kitty_exe():
|
||||
def kitty_exe() -> str:
|
||||
rpath = sys._xoptions.get('bundle_exe_dir')
|
||||
if not rpath:
|
||||
items = filter(None, os.environ.get('PATH', '').split(os.pathsep))
|
||||
@ -53,7 +63,7 @@ def kitty_exe():
|
||||
return os.path.join(rpath, 'kitty')
|
||||
|
||||
|
||||
def _get_config_dir():
|
||||
def _get_config_dir() -> str:
|
||||
if 'KITTY_CONFIG_DIRECTORY' in os.environ:
|
||||
return os.path.abspath(os.path.expanduser(os.environ['KITTY_CONFIG_DIRECTORY']))
|
||||
|
||||
@ -71,12 +81,12 @@ def _get_config_dir():
|
||||
if os.access(q, os.W_OK) and os.path.exists(os.path.join(q, 'kitty.conf')):
|
||||
return q
|
||||
|
||||
def make_tmp_conf():
|
||||
def make_tmp_conf() -> None:
|
||||
import tempfile
|
||||
import atexit
|
||||
ans = tempfile.mkdtemp(prefix='kitty-conf-')
|
||||
|
||||
def cleanup():
|
||||
def cleanup() -> None:
|
||||
import shutil
|
||||
with suppress(Exception):
|
||||
shutil.rmtree(ans)
|
||||
@ -103,7 +113,7 @@ defconf = os.path.join(config_dir, 'kitty.conf')
|
||||
|
||||
|
||||
@lru_cache(maxsize=2)
|
||||
def cache_dir():
|
||||
def cache_dir() -> str:
|
||||
if 'KITTY_CACHE_DIRECTORY' in os.environ:
|
||||
candidate = os.path.abspath(os.environ['KITTY_CACHE_DIRECTORY'])
|
||||
elif is_macos:
|
||||
@ -115,7 +125,7 @@ def cache_dir():
|
||||
return candidate
|
||||
|
||||
|
||||
def wakeup():
|
||||
def wakeup() -> None:
|
||||
from .fast_data_types import get_boss
|
||||
b = get_boss()
|
||||
if b is not None:
|
||||
@ -135,11 +145,11 @@ except KeyError:
|
||||
shell_path = '/bin/sh'
|
||||
|
||||
|
||||
def glfw_path(module):
|
||||
def glfw_path(module: str) -> str:
|
||||
return os.path.join(base, 'glfw-{}.so'.format(module))
|
||||
|
||||
|
||||
def detect_if_wayland_ok():
|
||||
def detect_if_wayland_ok() -> bool:
|
||||
if 'WAYLAND_DISPLAY' not in os.environ:
|
||||
return False
|
||||
if 'KITTY_DISABLE_WAYLAND' in os.environ:
|
||||
@ -169,11 +179,11 @@ def detect_if_wayland_ok():
|
||||
return ans == b'YES'
|
||||
|
||||
|
||||
def is_wayland(opts=None):
|
||||
def is_wayland(opts: Optional['Options'] = None) -> bool:
|
||||
if is_macos:
|
||||
return False
|
||||
if opts is None:
|
||||
return getattr(is_wayland, 'ans')
|
||||
return bool(getattr(is_wayland, 'ans'))
|
||||
if opts.linux_display_server == 'auto':
|
||||
ans = detect_if_wayland_ok()
|
||||
else:
|
||||
|
||||
@ -5,13 +5,12 @@
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
from collections import namedtuple
|
||||
from contextlib import suppress
|
||||
from typing import Optional
|
||||
from typing import Dict, NamedTuple, Optional
|
||||
from urllib.request import urlopen
|
||||
|
||||
from .config import atomic_save
|
||||
from .constants import cache_dir, kitty_exe, version
|
||||
from .constants import Version, cache_dir, kitty_exe, version
|
||||
from .fast_data_types import add_timer, get_boss, monitor_pid
|
||||
from .notify import notify
|
||||
from .utils import log_error, open_url
|
||||
@ -19,21 +18,26 @@ from .utils import log_error, open_url
|
||||
CHANGELOG_URL = 'https://sw.kovidgoyal.net/kitty/changelog.html'
|
||||
RELEASED_VERSION_URL = 'https://sw.kovidgoyal.net/kitty/current-version.txt'
|
||||
CHECK_INTERVAL = 24 * 60 * 60
|
||||
Notification = namedtuple('Notification', 'version time_of_last_notification count')
|
||||
|
||||
|
||||
def notification_activated():
|
||||
class Notification(NamedTuple):
|
||||
version: Version
|
||||
time_of_last_notification: float
|
||||
notification_count: int
|
||||
|
||||
|
||||
def notification_activated() -> None:
|
||||
open_url(CHANGELOG_URL)
|
||||
|
||||
|
||||
def version_notification_log():
|
||||
def version_notification_log() -> str:
|
||||
override = getattr(version_notification_log, 'override', None)
|
||||
if override:
|
||||
if isinstance(override, str):
|
||||
return override
|
||||
return os.path.join(cache_dir(), 'new-version-notifications-1.txt')
|
||||
|
||||
|
||||
def notify_new_version(release_version):
|
||||
def notify_new_version(release_version: Version) -> None:
|
||||
notify(
|
||||
'kitty update available!',
|
||||
'kitty version {} released'.format('.'.join(map(str, release_version))),
|
||||
@ -41,22 +45,23 @@ def notify_new_version(release_version):
|
||||
)
|
||||
|
||||
|
||||
def get_released_version():
|
||||
def get_released_version() -> str:
|
||||
try:
|
||||
raw = urlopen(RELEASED_VERSION_URL).read().decode('utf-8').strip()
|
||||
except Exception:
|
||||
raw = '0.0.0'
|
||||
return raw
|
||||
return str(raw)
|
||||
|
||||
|
||||
def parse_line(line):
|
||||
def parse_line(line: str) -> Notification:
|
||||
parts = line.split(',')
|
||||
version, timestamp, count = parts
|
||||
version = tuple(map(int, version.split('.')))
|
||||
return Notification(version, float(timestamp), int(count))
|
||||
parts = version.split('.')
|
||||
v = Version(int(parts[0]), int(parts[1]), int(parts[2]))
|
||||
return Notification(v, float(timestamp), int(count))
|
||||
|
||||
|
||||
def read_cache():
|
||||
def read_cache() -> Dict[Version, Notification]:
|
||||
notified_versions = {}
|
||||
with suppress(FileNotFoundError):
|
||||
with open(version_notification_log()) as f:
|
||||
@ -69,16 +74,16 @@ def read_cache():
|
||||
return notified_versions
|
||||
|
||||
|
||||
def already_notified(version):
|
||||
def already_notified(version: tuple) -> bool:
|
||||
notified_versions = read_cache()
|
||||
return version in notified_versions
|
||||
|
||||
|
||||
def save_notification(version):
|
||||
def save_notification(version: Version) -> None:
|
||||
notified_versions = read_cache()
|
||||
if version in notified_versions:
|
||||
v = notified_versions[version]
|
||||
notified_versions[version] = v._replace(time_of_last_notification=time.time(), count=v.count + 1)
|
||||
notified_versions[version] = v._replace(time_of_last_notification=time.time(), notification_count=v.notification_count + 1)
|
||||
else:
|
||||
notified_versions[version] = Notification(version, time.time(), 1)
|
||||
lines = []
|
||||
@ -89,8 +94,8 @@ def save_notification(version):
|
||||
atomic_save('\n'.join(lines).encode('utf-8'), version_notification_log())
|
||||
|
||||
|
||||
def process_current_release(raw):
|
||||
release_version = tuple(map(int, raw.split('.')))
|
||||
def process_current_release(raw: str) -> None:
|
||||
release_version = Version(*tuple(map(int, raw.split('.'))))
|
||||
if release_version > version and not already_notified(release_version):
|
||||
save_notification(release_version)
|
||||
notify_new_version(release_version)
|
||||
|
||||
@ -26,7 +26,7 @@ docs_dir = os.path.abspath('docs')
|
||||
publish_dir = os.path.abspath(os.path.join('..', 'kovidgoyal.github.io', 'kitty'))
|
||||
with open('kitty/constants.py') as f:
|
||||
raw = f.read()
|
||||
nv = re.search(r'^version\s+=\s+\((\d+), (\d+), (\d+)\)', raw, flags=re.MULTILINE)
|
||||
nv = re.search(r'^version\s+=\s+Version\((\d+), (\d+), (\d+)\)', raw, flags=re.MULTILINE)
|
||||
if nv is not None:
|
||||
version = '%s.%s.%s' % (nv.group(1), nv.group(2), nv.group(3))
|
||||
appname = re.search(r"^appname\s+=\s+'([^']+)'", raw, flags=re.MULTILINE).group(1) # type: ignore
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user