# -*- coding: utf-8 -*-
"""
Classes used both by front-end and back-end
"""
import os.path
import site
import sys
from collections import namedtuple
from dataclasses import dataclass
from logging import getLogger
from typing import Any, Callable, Dict, List, Optional, Tuple # @UnusedImport
logger = getLogger(__name__)
STRING_PSEUDO_FILENAME = "<string>"
REPL_PSEUDO_FILENAME = "<stdin>"
MESSAGE_MARKER = "\x02"
OBJECT_LINK_START = "[object_link_for_thonny=%d]"
OBJECT_LINK_END = "[/object_link_for_thonny]"
REMOTE_PATH_MARKER = " :: "
PROCESS_ACK = "OK"
IGNORED_FILES_AND_DIRS = [
"System Volume Information",
"._.Trashes",
".Trashes",
"__MACOSX",
".DS_Store",
]
ValueInfo = namedtuple("ValueInfo", ["id", "repr"])
FrameInfo = namedtuple(
"FrameInfo",
[
"id",
"filename",
"module_name",
"code_name",
"source",
"lineno",
"firstlineno",
"in_library",
"locals",
"globals",
"freevars",
"event",
"focus",
"node_tags",
"current_statement",
"current_root_expression",
"current_evaluations",
],
)
TextRange = namedtuple("TextRange", ["lineno", "col_offset", "end_lineno", "end_col_offset"])
@dataclass(frozen=True)
class DistInfo:
key: str
project_name: str
version: str
location: str
class Record:
def __init__(self, **kw):
self.__dict__.update(kw)
def update(self, e, **kw):
self.__dict__.update(e, **kw)
def setdefault(self, **kw):
"updates those fields that are not yet present (similar to dict.setdefault)"
for key in kw:
if not hasattr(self, key):
setattr(self, key, kw[key])
def get(self, key, default=None):
return self.__dict__.get(key, default)
def __getitem__(self, key):
return self.__dict__[key]
def __delitem__(self, key):
self.__dict__.__delitem__(key)
def __setitem__(self, key, value):
self.__dict__[key] = value
def __contains__(self, key):
return key in self.__dict__
def __repr__(self):
keys = self.__dict__.keys()
items = ("{}={}".format(k, repr(self.__dict__[k])) for k in keys)
return "{}({})".format(self.__class__.__name__, ", ".join(items))
def __str__(self):
keys = sorted(self.__dict__.keys())
items = ("{}={}".format(k, repr(self.__dict__[k])) for k in keys)
return "{}({})".format(self.__class__.__name__, ", ".join(items))
def __eq__(self, other):
# pylint: disable=unidiomatic-typecheck
if type(self) != type(other):
return False
if len(self.__dict__) != len(other.__dict__):
return False
for key in self.__dict__:
if not hasattr(other, key):
return False
self_value = getattr(self, key)
other_value = getattr(other, key)
if type(self_value) != type(other_value) or self_value != other_value:
return False
return True
def __ne__(self, other):
return not self.__eq__(other)
def __hash__(self):
return hash(repr(self))
def range_contains_smaller(one: TextRange, other: TextRange) -> bool:
this_start = (one.lineno, one.col_offset)
this_end = (one.end_lineno, one.end_col_offset)
other_start = (other.lineno, other.col_offset)
other_end = (other.end_lineno, other.end_col_offset)
return (
this_start < other_start
and this_end > other_end
or this_start == other_start
and this_end > other_end
or this_start < other_start
and this_end == other_end
)
def range_contains_smaller_or_equal(one: TextRange, other: TextRange) -> bool:
return range_contains_smaller(one, other) or one == other
class InputSubmission(Record):
"""For sending data to backend's stdin"""
def __init__(self, data: str, **kw) -> None:
super().__init__(**kw)
self.data = data
class CommandToBackend(Record):
"""Command meant for the back-end"""
def __init__(self, name: str, **kw) -> None:
super().__init__(**kw)
self.name = name
class ImmediateCommand(CommandToBackend):
pass
class EOFCommand(CommandToBackend):
def __init__(self, **kw) -> None:
if "name" in kw:
del kw["name"]
super().__init__("eof", **kw)
class ToplevelCommand(CommandToBackend):
def __init__(self, name: str, argv: List[str] = [], **kw) -> None:
super().__init__(name, **kw)
self.argv = argv
class DebuggerCommand(CommandToBackend):
pass
class InlineCommand(CommandToBackend):
"""
Can be used both during debugging and in waiting_toplevel_command state
(eg. for sending variable and heap info requests)
"""
pass
class MessageFromBackend(Record):
def __init__(self, **kw) -> None:
self.event_type = type(self).__name__ # allow event_type to be overridden by kw
super().__init__(**kw)
if not hasattr(self, "sequence"):
self.sequence = self.event_type
class ToplevelResponse(MessageFromBackend):
pass
class DebuggerResponse(MessageFromBackend):
pass
class BackendEvent(MessageFromBackend):
def __init__(self, event_type: str, **kw) -> None:
super().__init__(**kw)
self.event_type = event_type
class OscEvent(BackendEvent):
def __init__(self, text: str):
self.text = text
super().__init__("OscEvent")
def __repr__(self):
return f"OscEvent({self.text!r})"
class InlineResponse(MessageFromBackend):
def __init__(self, command_name: str, **kw) -> None:
super().__init__(**kw)
self.command_name = command_name
self.event_type = self.command_name + "_response"
def serialize_message(msg: Record, max_line_length=65536) -> str:
# I want to transfer only ASCII chars because encodings are not reliable
# (eg. can't find a way to specify PYTHONIOENCODING for cx_freeze'd program)
# The possibility for splitting message into several lines is required because of
# default (safe) window size in Paramiko (https://github.com/thonny/thonny/issues/1680)
msg_str = ascii(msg)
lines = []
for i in range(0, len(msg_str), max_line_length):
lines.append(msg_str[i : i + max_line_length])
return MESSAGE_MARKER + str(len(lines)) + " " + "\n".join(lines)
def parse_message(msg_string: str) -> Record:
# DataFrames may have nan
# pylint: disable=unused-variable
nan = float("nan") # @UnusedVariable
assert msg_string[0] == MESSAGE_MARKER
assert msg_string.strip().endswith(")")
msg_start = msg_string.index(" ")
line_count = int(msg_string[1:msg_start])
assert line_count == msg_string.strip().count("\n") + 1
return eval(msg_string[msg_start:].replace("\n", ""))
def normpath_with_actual_case(name: str) -> str:
"""In Windows return the path with the case it is stored in the filesystem"""
if not os.path.exists(name):
return os.path.normpath(name)
if os.name == "nt":
try:
# https://stackoverflow.com/questions/2113822/python-getting-filename-case-as-stored-in-windows/2114975
norm_name = os.path.normpath(name)
from ctypes import create_unicode_buffer, windll
buf = create_unicode_buffer(512)
# GetLongPathNameW alone doesn't fix filename part
windll.kernel32.GetShortPathNameW(norm_name, buf, 512) # @UndefinedVariable
windll.kernel32.GetLongPathNameW(buf.value, buf, 512) # @UndefinedVariable
result = buf.value
if result.casefold() != norm_name.casefold():
# Sometimes GetShortPathNameW + GetLongPathNameW doesn't work
# see eg. https://github.com/thonny/thonny/issues/925
windll.kernel32.GetLongPathNameW(norm_name, buf, 512) # @UndefinedVariable
result = buf.value
if result.casefold() != norm_name.casefold():
result = norm_name
if result[1] == ":":
# ensure drive letter is capital
return result[0].upper() + result[1:]
else:
return result
except Exception:
logger.warning(
"Could not compute normpath_with_actual_case for %r", name, exc_info=True
)
return os.path.normpath(name)
else:
# easy on Linux
# too difficult on mac
# https://stackoverflow.com/questions/14515073/in-python-on-osx-with-hfs-how-can-i-get-the-correct-case-of-an-existing-filenam
# Hopefully only correct case comes into Thonny (eg. via open dialog)
return os.path.normpath(name)
def is_same_path(name1: str, name2: str) -> bool:
return os.path.normpath(os.path.normcase(name1)) == os.path.normpath(os.path.normcase(name2))
def path_startswith(child_name: str, dir_name: str) -> bool:
normchild = os.path.normpath(os.path.normcase(child_name))
normdir = os.path.normpath(os.path.normcase(dir_name))
return normdir == normchild or normchild.startswith(normdir.rstrip(os.path.sep) + os.path.sep)
def read_source(filename):
import tokenize
with tokenize.open(filename) as fp:
return fp.read()
def get_exe_dirs():
result = []
if site.ENABLE_USER_SITE:
if sys.platform == "win32":
if site.getusersitepackages():
result.append(site.getusersitepackages().replace("site-packages", "Scripts"))
else:
if site.getuserbase():
result.append(site.getuserbase() + "/bin")
main_scripts = os.path.join(sys.prefix, "Scripts")
if os.path.isdir(main_scripts) and main_scripts not in result:
result.append(main_scripts)
if os.path.dirname(sys.executable) not in result:
result.append(os.path.dirname(sys.executable))
# These entries are used by Anaconda
for part in [
"Library/mingw-w64/bin",
"Library/usr/bin",
"Library/bin",
"Scripts",
"bin",
"condabin",
]:
dirpath = os.path.join(sys.prefix, part.replace("/", os.sep))
if os.path.isdir(dirpath) and dirpath not in result:
result.append(dirpath)
if sys.platform != "win32" and "/usr/local/bin" not in result:
# May be missing on macOS, when started as bundle
# (yes, more may be missing, but this one is most useful)
result.append("/usr/local/bin")
return result
def get_site_dir(symbolic_name, executable=None):
if not executable or executable == sys.executable:
result = getattr(site, symbolic_name, "")
else:
import subprocess
result = (
subprocess.check_output(
[executable, "-m", "site", "--" + symbolic_name.lower().replace("_", "-")],
universal_newlines=True,
)
.decode()
.strip()
)
return result if result else None
def get_base_executable():
if sys.exec_prefix == sys.base_exec_prefix:
return sys.executable
if sys.platform == "win32":
guess = sys.base_exec_prefix + "\\" + os.path.basename(sys.executable)
if os.path.isfile(guess):
return normpath_with_actual_case(guess)
if os.path.islink(sys.executable):
return os.path.realpath(sys.executable)
raise RuntimeError("Don't know how to locate base executable")
def get_augmented_system_path(extra_dirs):
path_items = os.environ.get("PATH", "").split(os.pathsep)
for d in reversed(extra_dirs):
if d not in path_items:
path_items.insert(0, d)
return os.pathsep.join(path_items)
def update_system_path(env, value):
# in Windows, env keys are not case sensitive
# this is important if env is a dict (not os.environ)
if sys.platform == "win32":
found = False
for key in env:
if key.upper() == "PATH":
found = True
env[key] = value
if not found:
env["PATH"] = value
else:
env["PATH"] = value
@dataclass
class SignatureParameter:
kind: str
name: str
annotation: Optional[str]
default: Optional[str]
@dataclass
class SignatureInfo:
name: str
params: List[SignatureParameter]
return_type: Optional[str]
current_param_index: Optional[int] = None
call_bracket_start: Optional[Tuple[int, int]] = None
@dataclass
class CompletionInfo:
name: str
name_with_symbols: str
full_name: str
type: str
prefix_length: int # the number of chars to be deleted before inserting name
signatures: Optional[List[SignatureInfo]]
docstring: Optional[str]
module_name: Optional[str]
module_path: Optional[str]
@dataclass
class NameReference:
module_name: str
module_path: str
row: int
column: int
length: int
class UserError(RuntimeError):
"""Errors of this class are meant to be presented without stacktrace"""
pass
def is_hidden_or_system_file(path: str) -> bool:
if os.path.basename(path).startswith("."):
return True
elif sys.platform == "win32":
from ctypes import windll
FILE_ATTRIBUTE_HIDDEN = 0x2
FILE_ATTRIBUTE_SYSTEM = 0x4
return bool(
windll.kernel32.GetFileAttributesW(path) # @UndefinedVariable
& (FILE_ATTRIBUTE_HIDDEN | FILE_ATTRIBUTE_SYSTEM)
)
else:
return False
def get_dirs_children_info(
paths: List[str], include_hidden: bool = False
) -> Dict[str, Optional[Dict[str, Dict]]]:
return {path: get_single_dir_child_data(path, include_hidden) for path in paths}
def get_single_dir_child_data(path: str, include_hidden: bool = False) -> Optional[Dict[str, Dict]]:
if path == "":
if sys.platform == "win32":
return {**get_windows_volumes_info(), **get_windows_network_locations()}
else:
return get_single_dir_child_data("/", include_hidden)
elif os.path.isdir(path) or os.path.ismount(path):
result = {}
try:
for child in os.listdir(path):
full_child_path = os.path.join(path, child)
if not os.path.exists(full_child_path):
# must be broken link
continue
full_child_path = normpath_with_actual_case(full_child_path)
hidden = is_hidden_or_system_file(full_child_path)
if not hidden or include_hidden:
name = os.path.basename(full_child_path)
st = os.stat(full_child_path, dir_fd=None, follow_symlinks=True)
result[name] = {
"size": None if os.path.isdir(full_child_path) else st.st_size,
"modified": st.st_mtime,
"hidden": hidden,
}
except PermissionError:
result["<not accessible>"] = {
"kind": "error",
"size": -1,
"modified": None,
"hidden": None,
}
return result
else:
return None
def get_windows_volumes_info():
# http://stackoverflow.com/a/2288225/261181
# http://msdn.microsoft.com/en-us/library/windows/desktop/aa364939%28v=vs.85%29.aspx
import string
from ctypes import windll
all_drive_types = [
"DRIVE_UNKNOWN",
"DRIVE_NO_ROOT_DIR",
"DRIVE_REMOVABLE",
"DRIVE_FIXED",
"DRIVE_REMOTE",
"DRIVE_CDROM",
"DRIVE_RAMDISK",
]
required_drive_types = ["DRIVE_REMOVABLE", "DRIVE_FIXED", "DRIVE_REMOTE", "DRIVE_RAMDISK"]
result = {}
bitmask = windll.kernel32.GetLogicalDrives() # @UndefinedVariable
for letter in string.ascii_uppercase:
if not bitmask & 1:
pass
else:
drive_type = all_drive_types[
windll.kernel32.GetDriveTypeW("%s:\\" % letter)
] # @UndefinedVariable
# NB! Drive A can be present in bitmask but actually missing.
# In this case querying information about it would freeze the UI
# for several seconds.
# One solution is to uninstall the device in device manager,
# but OS may restore the drive later.
# Therefore it is safest to skip A drive (user can access it via Open dialog)
if drive_type in required_drive_types and (
letter != "A" or drive_type != "DRIVE_REMOVABLE"
):
drive = letter + ":"
path = drive + "\\"
try:
st = os.stat(path)
volume_name = get_windows_volume_name(path)
if not volume_name:
volume_name = "Disk"
label = volume_name + " (" + drive + ")"
result[path] = {
"label": label,
"size": None,
"modified": max(st.st_mtime, st.st_ctime),
}
except OSError as e:
logger.warning("Could not get information for %s", path, exc_info=e)
bitmask >>= 1
return result
def get_windows_volume_name(path):
# https://stackoverflow.com/a/12056414/261181
import ctypes
kernel32 = ctypes.windll.kernel32
volume_name_buffer = ctypes.create_unicode_buffer(1024)
file_system_name_buffer = ctypes.create_unicode_buffer(1024)
serial_number = None
max_component_length = None
file_system_flags = None
result = kernel32.GetVolumeInformationW(
ctypes.c_wchar_p(path),
volume_name_buffer,
ctypes.sizeof(volume_name_buffer),
serial_number,
max_component_length,
file_system_flags,
file_system_name_buffer,
ctypes.sizeof(file_system_name_buffer),
)
if result:
return volume_name_buffer.value
else:
return None
def get_windows_network_locations():
import ctypes.wintypes
CSIDL_NETHOOD = 0x13
SHGFP_TYPE_CURRENT = 0
buf = ctypes.create_unicode_buffer(ctypes.wintypes.MAX_PATH)
ctypes.windll.shell32.SHGetFolderPathW(0, CSIDL_NETHOOD, 0, SHGFP_TYPE_CURRENT, buf)
shortcuts_dir = buf.value
if not buf.value:
logger.warning("Could not determine windows shortcuts directory")
return {}
result = {}
for entry in os.scandir(shortcuts_dir):
# full_path = normpath_with_actual_case(entry.path)
lnk_path = os.path.join(entry.path, "target.lnk")
if os.path.exists(lnk_path):
try:
target = get_windows_lnk_target(lnk_path)
result[target] = {
"label": entry.name + " (" + target + ")",
"size": None,
"modified": None,
}
except Exception:
logger.error("Can't get target from %s", lnk_path, exc_info=True)
return result
def get_windows_lnk_target(lnk_file_path):
import subprocess
import thonny
script_path = os.path.join(os.path.dirname(thonny.__file__), "res", "PrintLnkTarget.vbs")
cmd = ["cscript", "/NoLogo", script_path, lnk_file_path]
result = subprocess.check_output(cmd, universal_newlines=True, timeout=3)
return result.strip()
def execute_system_command(cmd, cwd=None, disconnect_stdin=False):
import subprocess
logger.debug("execute_system_command, cmd=%r, cwd=%s", cmd, cwd)
env = dict(os.environ).copy()
encoding = "utf-8"
env["PYTHONIOENCODING"] = encoding
# Make sure this python interpreter and its scripts are available
# in PATH
update_system_path(env, get_augmented_system_path(get_exe_dirs()))
popen_kw = dict(
env=env,
universal_newlines=True,
bufsize=0,
)
if cwd and os.path.isdir(cwd):
popen_kw["cwd"] = cwd
if disconnect_stdin:
popen_kw["stdin"] = subprocess.DEVNULL
if sys.version_info >= (3, 6):
popen_kw["errors"] = "replace"
popen_kw["encoding"] = encoding
if isinstance(cmd.cmd_line, str) and cmd.cmd_line.startswith("!"):
cmd_line = cmd.cmd_line[1:]
popen_kw["shell"] = True
else:
assert isinstance(cmd.cmd_line, list)
cmd_line = cmd.cmd_line
logger.debug("Popen(%r, ...)", cmd_line)
proc = subprocess.Popen(cmd_line, **popen_kw)
proc.communicate()
return proc.wait()
def universal_dirname(path: str) -> str:
if "/" in path:
sep = "/"
elif "\\" in path:
sep = "\\"
else:
# micro:bit
return ""
path = path.rstrip(sep)
result = path[: path.rindex(sep)]
if not result:
return sep
else:
return result
def universal_relpath(path: str, context: str) -> str:
"""Tries to give relative path"""
if "/" in path:
import pathlib
p = pathlib.PurePosixPath(path)
try:
return str(p.relative_to(context))
except ValueError:
return path
else:
return os.path.relpath(path, context)
def get_python_version_string():
result = sys.version.split()[0]
if sys.maxsize <= 2**32:
result += ", 32-bit"
return result
def execute_with_frontend_sys_path(function: Callable) -> Any:
import ast
try:
frontend_sys_path = ast.literal_eval(os.environ["THONNY_FRONTEND_SYS_PATH"])
assert isinstance(frontend_sys_path, list)
logger.info("Using THONNY_FRONTEND_SYS_PATH %s", frontend_sys_path)
except Exception as e:
logger.debug("Could not get THONNY_FRONTEND_SYS_PATH", exc_info=e)
frontend_sys_path = []
extra_items = [item for item in frontend_sys_path if item not in sys.path]
sys.path = sys.path + extra_items
try:
return function()
finally:
for item in extra_items:
if item in sys.path:
sys.path.remove(item)
def try_load_modules_with_frontend_sys_path(module_names):
def load():
from importlib import import_module
for name in module_names:
try:
import_module(name)
except ImportError:
pass
execute_with_frontend_sys_path(load)
def read_one_incoming_message_str(line_reader):
msg_str = line_reader()
if msg_str == "":
return ""
if not msg_str.startswith(MESSAGE_MARKER):
return msg_str
line_count = int(msg_str[1:].split(maxsplit=1)[0])
read_lines = 1
while read_lines < line_count:
msg_str += line_reader()
read_lines += 1
return msg_str
def is_virtual_executable(executable):
exe_dir = os.path.dirname(executable)
return os.path.exists(os.path.join(exe_dir, "activate")) or os.path.exists(
os.path.join(exe_dir, "activate.bat")
)
def is_private_python(executable):
result = os.path.exists(os.path.join(os.path.dirname(executable), "thonny_python.ini"))
logger.debug("is_private_python(%r) == %r", executable, result)
return result
def running_in_virtual_environment() -> bool:
return (
hasattr(sys, "base_prefix")
and sys.base_prefix != sys.prefix
or hasattr(sys, "real_prefix")
and getattr(sys, "real_prefix") != sys.prefix
)
def is_remote_path(s: str) -> bool:
return REMOTE_PATH_MARKER in s
def is_local_path(s: str) -> bool:
return not is_remote_path(s) and not s.startswith("<")