vscode-python/pythonFiles/PythonTools/visualstudio_py_repl.py at master · hlinet/vscode-python · GitHub

History

1391 lines (1161 loc) · 48.8 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

653

654

655

656

657

658

659

660

661

662

663

664

665

666

667

668

669

670

671

672

673

674

675

676

677

678

679

680

681

682

683

684

685

686

687

688

689

690

691

692

693

694

695

696

697

698

699

700

701

702

703

704

705

706

707

708

709

710

711

712

713

714

715

716

717

718

719

720

721

722

723

724

725

726

727

728

729

730

731

732

733

734

735

736

737

738

739

740

741

742

743

744

745

746

747

748

749

750

751

752

753

754

755

756

757

758

759

760

761

762

763

764

765

766

767

768

769

770

771

772

773

774

775

776

777

778

779

780

781

782

783

784

785

786

787

788

789

790

791

792

793

794

795

796

797

798

799

800

801

802

803

804

805

806

807

808

809

810

811

812

813

814

815

816

817

818

819

820

821

822

823

824

825

826

827

828

829

830

831

832

833

834

835

836

837

838

839

840

841

842

843

844

845

846

847

848

849

850

851

852

853

854

855

856

857

858

859

860

861

862

863

864

865

866

867

868

869

870

871

872

873

874

875

876

877

878

879

880

881

882

883

884

885

886

887

888

889

890

891

892

893

894

895

896

897

898

899

900

901

902

903

904

905

906

907

908

909

910

911

912

913

914

915

916

917

918

919

920

921

922

923

924

925

926

927

928

929

930

931

932

933

934

935

936

937

938

939

940

941

942

943

944

945

946

947

948

949

950

951

952

953

954

955

956

957

958

959

960

961

962

963

964

965

966

967

968

969

970

971

972

973

974

975

976

977

978

979

980

981

982

983

984

985

986

987

988

989

990

991

992

993

994

995

996

997

998

999

1000

# Python Tools for Visual Studio

# Copyright(c) Microsoft Corporation

# Licensed under the Apache License, Version 2.0 (the License); you may not use

# this file except in compliance with the License. You may obtain a copy of the

# License at http://www.apache.org/licenses/LICENSE-2.0

# THIS CODE IS PROVIDED ON AN *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS

# OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY

# IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE,

# MERCHANTABLITY OR NON-INFRINGEMENT.

# See the Apache Version 2.0 License for specific language governing

# permissions and limitations under the License.

from __future__ import with_statement

__author__ = "Microsoft Corporation <ptvshelp@microsoft.com>"

__version__ = "3.0.0.0"

# This module MUST NOT import threading in global scope. This is because in a direct (non-ptvsd)

# attach scenario, it is loaded on the injected debugger attach thread, and if threading module

# hasn't been loaded already, it will assume that the thread on which it is being loaded is the

# main thread. This will cause issues when the thread goes away after attach completes.

try:

import thread

except ImportError:

# Renamed in Python3k

import _thread as thread

try:

from ssl import SSLError

except:

SSLError = None

import sys

import socket

import select

import time

import struct

import imp

import traceback

import random

import os

import inspect

import types

from collections import deque

try:

# In the local attach scenario, visualstudio_py_util is injected into globals()

# by PyDebugAttach before loading this module, and cannot be imported.

_vspu = visualstudio_py_util

except:

try:

import visualstudio_py_util as _vspu

except ImportError:

import ptvsd.visualstudio_py_util as _vspu

to_bytes = _vspu.to_bytes

read_bytes = _vspu.read_bytes

read_int = _vspu.read_int

read_string = _vspu.read_string

write_bytes = _vspu.write_bytes

write_int = _vspu.write_int

write_string = _vspu.write_string

try:

unicode

except NameError:

unicode = str

try:

BaseException

except NameError:

# BaseException not defined until Python 2.5

BaseException = Exception

DEBUG = os.environ.get('DEBUG_REPL') is not None

__all__ = ['ReplBackend', 'BasicReplBackend', 'BACKEND']

def _debug_write(out):

if DEBUG:

sys.__stdout__.write(out)

sys.__stdout__.flush()

class SafeSendLock(object):

"""a lock which ensures we're released if we take a KeyboardInterrupt exception acquiring it"""

def __init__(self):

self.lock = thread.allocate_lock()

def __enter__(self):

self.acquire()

def __exit__(self, exc_type, exc_value, tb):

self.release()

def acquire(self):

try:

self.lock.acquire()

except KeyboardInterrupt:

try:

self.lock.release()

except:

pass

raise

def release(self):

self.lock.release()

def _command_line_to_args_list(cmdline):

"""splits a string into a list using Windows command line syntax."""

args_list = []

if cmdline and cmdline.strip():

from ctypes import c_int, c_voidp, c_wchar_p

from ctypes import byref, POINTER, WinDLL

clta = WinDLL('shell32').CommandLineToArgvW

clta.argtypes = [c_wchar_p, POINTER(c_int)]

clta.restype = POINTER(c_wchar_p)

lf = WinDLL('kernel32').LocalFree

lf.argtypes = [c_voidp]

pNumArgs = c_int()

r = clta(cmdline, byref(pNumArgs))

if r:

for index in range(0, pNumArgs.value):

if sys.hexversion >= 0x030000F0:

argval = r[index]

else:

argval = r[index].encode('ascii', 'replace')

args_list.append(argval)

lf(r)

else:

sys.stderr.write('Error parsing script arguments:\n')

sys.stderr.write(cmdline + '\n')

return args_list

class UnsupportedReplException(Exception):

def __init__(self, reason):

self.reason = reason

# save the start_new_thread so we won't debug/break into the REPL comm thread.

start_new_thread = thread.start_new_thread

class ReplBackend(object):

"""back end for executing REPL code. This base class handles all of the

communication with the remote process while derived classes implement the

actual inspection and introspection."""

_MRES = to_bytes('MRES')

_SRES = to_bytes('SRES')

_MODS = to_bytes('MODS')

_IMGD = to_bytes('IMGD')

_PRPC = to_bytes('PRPC')

_RDLN = to_bytes('RDLN')

_STDO = to_bytes('STDO')

_STDE = to_bytes('STDE')

_DBGA = to_bytes('DBGA')

_DETC = to_bytes('DETC')

_DPNG = to_bytes('DPNG')

_DXAM = to_bytes('DXAM')

_CHWD = to_bytes('CHWD')

_MERR = to_bytes('MERR')

_SERR = to_bytes('SERR')

_ERRE = to_bytes('ERRE')

_EXIT = to_bytes('EXIT')

_DONE = to_bytes('DONE')

_MODC = to_bytes('MODC')

def __init__(self, *args, **kwargs):

import threading

self.conn = None

self.send_lock = SafeSendLock()

self.input_event = threading.Lock()

self.input_event.acquire() # lock starts acquired (we use it like a manual reset event)

self.input_string = None

self.exit_requested = False

def connect(self, port):

self.conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

self.conn.connect(('127.0.0.1', port))

# start a new thread for communicating w/ the remote process

start_new_thread(self._repl_loop, ())

def connect_using_socket(self, socket):

self.conn = socket

start_new_thread(self._repl_loop, ())

def _repl_loop(self):

"""loop on created thread which processes communicates with the REPL window"""

try:

while True:

if self.check_for_exit_repl_loop():

break

# we receive a series of 4 byte commands. Each command then

# has it's own format which we must parse before continuing to

# the next command.

self.flush()

self.conn.settimeout(10)

# 2.x raises SSLError in case of timeout (http://bugs.python.org/issue10272)

if SSLError:

timeout_exc_types = (socket.timeout, SSLError)

else:

timeout_exc_types = socket.timeout

try:

inp = read_bytes(self.conn, 4)

except timeout_exc_types:

r, w, x = select.select([], [], [self.conn], 0)

if x:

# an exception event has occured on the socket...

raise

continue

self.conn.settimeout(None)

if inp == '':

break

self.flush()

cmd = ReplBackend._COMMANDS.get(inp)

if cmd is not None:

cmd(self)

except:

_debug_write('error in repl loop')

_debug_write(traceback.format_exc())

self.exit_process()

time.sleep(2) # try and exit gracefully, then interrupt main if necessary

if sys.platform == 'cli':

# just kill us as fast as possible

import System

System.Environment.Exit(1)

self.interrupt_main()

def check_for_exit_repl_loop(self):

return False

def _cmd_run(self):

"""runs the received snippet of code"""

self.run_command(read_string(self.conn))

def _cmd_abrt(self):

"""aborts the current running command"""

# abort command, interrupts execution of the main thread.

self.interrupt_main()

def _cmd_exit(self):

"""exits the interactive process"""

self.exit_requested = True

self.exit_process()

def _cmd_mems(self):

"""gets the list of members available for the given expression"""

expression = read_string(self.conn)

try:

name, inst_members, type_members = self.get_members(expression)

except:

with self.send_lock:

write_bytes(self.conn, ReplBackend._MERR)

_debug_write('error in eval')

_debug_write(traceback.format_exc())

else:

with self.send_lock:

write_bytes(self.conn, ReplBackend._MRES)

write_string(self.conn, name)

self._write_member_dict(inst_members)

self._write_member_dict(type_members)

def _cmd_sigs(self):

"""gets the signatures for the given expression"""

expression = read_string(self.conn)

try:

sigs = self.get_signatures(expression)

except:

with self.send_lock:

write_bytes(self.conn, ReplBackend._SERR)

_debug_write('error in eval')

_debug_write(traceback.format_exc())

else:

with self.send_lock:

write_bytes(self.conn, ReplBackend._SRES)

# single overload

write_int(self.conn, len(sigs))

for doc, args, vargs, varkw, defaults in sigs:

# write overload

write_string(self.conn, (doc or '')[:4096])

arg_count = len(args) + (vargs is not None) + (varkw is not None)

write_int(self.conn, arg_count)

def_values = [''] * (len(args) - len(defaults)) + ['=' + d for d in defaults]

for arg, def_value in zip(args, def_values):

write_string(self.conn, (arg or '') + def_value)

if vargs is not None:

write_string(self.conn, '*' + vargs)

if varkw is not None:

write_string(self.conn, '**' + varkw)

def _cmd_setm(self):

global exec_mod

"""sets the current module which code will execute against"""

mod_name = read_string(self.conn)

self.set_current_module(mod_name)

def _cmd_sett(self):

"""sets the current thread and frame which code will execute against"""

thread_id = read_int(self.conn)

frame_id = read_int(self.conn)

frame_kind = read_int(self.conn)

self.set_current_thread_and_frame(thread_id, frame_id, frame_kind)

def _cmd_mods(self):

"""gets the list of available modules"""

try:

res = self.get_module_names()

res.sort()

except:

res = []

with self.send_lock:

write_bytes(self.conn, ReplBackend._MODS)

write_int(self.conn, len(res))

for name, filename in res:

write_string(self.conn, name)

write_string(self.conn, filename)

def _cmd_inpl(self):

"""handles the input command which returns a string of input"""

self.input_string = read_string(self.conn)

self.input_event.release()

def _cmd_excf(self):

"""handles executing a single file"""

filename = read_string(self.conn)

args = read_string(self.conn)

self.execute_file(filename, args)

def _cmd_excx(self):

"""handles executing a single file, module or process"""

filetype = read_string(self.conn)

filename = read_string(self.conn)

args = read_string(self.conn)

self.execute_file_ex(filetype, filename, args)

def _cmd_debug_attach(self):

import visualstudio_py_debugger

port = read_int(self.conn)

id = read_string(self.conn)

debug_options = visualstudio_py_debugger.parse_debug_options(read_string(self.conn))

self.attach_process(port, id, debug_options)

_COMMANDS = {

to_bytes('run '): _cmd_run,

to_bytes('abrt'): _cmd_abrt,

to_bytes('exit'): _cmd_exit,

to_bytes('mems'): _cmd_mems,

to_bytes('sigs'): _cmd_sigs,

to_bytes('mods'): _cmd_mods,

to_bytes('setm'): _cmd_setm,

to_bytes('sett'): _cmd_sett,

to_bytes('inpl'): _cmd_inpl,

to_bytes('excf'): _cmd_excf,

to_bytes('excx'): _cmd_excx,

to_bytes('dbga'): _cmd_debug_attach,

}

def _write_member_dict(self, mem_dict):

write_int(self.conn, len(mem_dict))

for name, type_name in mem_dict.items():

write_string(self.conn, name)

write_string(self.conn, type_name)

def on_debugger_detach(self):

with self.send_lock:

write_bytes(self.conn, ReplBackend._DETC)

def init_debugger(self):

from os import path

sys.path.append(path.dirname(__file__))

import visualstudio_py_debugger

visualstudio_py_debugger.DONT_DEBUG.append(path.normcase(__file__))

new_thread = visualstudio_py_debugger.new_thread()

sys.settrace(new_thread.trace_func)

visualstudio_py_debugger.intercept_threads(True)

def send_image(self, filename):

with self.send_lock:

write_bytes(self.conn, ReplBackend._IMGD)

write_string(self.conn, filename)

def write_png(self, image_bytes):

with self.send_lock:

write_bytes(self.conn, ReplBackend._DPNG)

write_int(self.conn, len(image_bytes))

write_bytes(self.conn, image_bytes)

def write_xaml(self, xaml_bytes):

with self.send_lock:

write_bytes(self.conn, ReplBackend._DXAM)

write_int(self.conn, len(xaml_bytes))

write_bytes(self.conn, xaml_bytes)

def send_prompt(self, ps1, ps2, allow_multiple_statements):

"""sends the current prompt to the interactive window"""

with self.send_lock:

write_bytes(self.conn, ReplBackend._PRPC)

write_string(self.conn, ps1)

write_string(self.conn, ps2)

write_int(self.conn, 1 if allow_multiple_statements else 0)

def send_cwd(self):

"""sends the current working directory"""

with self.send_lock:

write_bytes(self.conn, ReplBackend._CHWD)

write_string(self.conn, os.getcwd())

def send_error(self):

"""reports that an error occured to the interactive window"""

with self.send_lock:

write_bytes(self.conn, ReplBackend._ERRE)

def send_exit(self):

"""reports the that the REPL process has exited to the interactive window"""

with self.send_lock:

write_bytes(self.conn, ReplBackend._EXIT)

def send_command_executed(self):

with self.send_lock:

write_bytes(self.conn, ReplBackend._DONE)

def send_modules_changed(self):

with self.send_lock:

write_bytes(self.conn, ReplBackend._MODC)

def read_line(self):

"""reads a line of input from standard input"""

with self.send_lock:

write_bytes(self.conn, ReplBackend._RDLN)

self.input_event.acquire()

return self.input_string

def write_stdout(self, value):

"""writes a string to standard output in the remote console"""

with self.send_lock:

write_bytes(self.conn, ReplBackend._STDO)

write_string(self.conn, value)

def write_stderr(self, value):

"""writes a string to standard input in the remote console"""

with self.send_lock:

write_bytes(self.conn, ReplBackend._STDE)

write_string(self.conn, value)

################################################################

# Implementation of execution, etc...

def execution_loop(self):

"""starts processing execution requests"""

raise NotImplementedError

def run_command(self, command):

"""runs the specified command which is a string containing code"""

raise NotImplementedError

def execute_file(self, filename, args):

"""executes the given filename as the main module"""

return self.execute_file_ex('script', filename, args)

def execute_file_ex(self, filetype, filename, args):

"""executes the given filename as a 'script', 'module' or 'process'."""

raise NotImplementedError

def interrupt_main(self):

"""aborts the current running command"""

raise NotImplementedError

def exit_process(self):

"""exits the REPL process"""

raise NotImplementedError

def get_members(self, expression):

"""returns a tuple of the type name, instance members, and type members"""

raise NotImplementedError

def get_signatures(self, expression):

"""returns doc, args, vargs, varkw, defaults."""

raise NotImplementedError

def set_current_module(self, module):

"""sets the module which code executes against"""

raise NotImplementedError

def set_current_thread_and_frame(self, thread_id, frame_id, frame_kind):

"""sets the current thread and frame which code will execute against"""

raise NotImplementedError

def get_module_names(self):

"""returns a list of module names"""

raise NotImplementedError

def flush(self):

"""flushes the stdout/stderr buffers"""

raise NotImplementedError

def attach_process(self, port, debugger_id, debug_options):

"""starts processing execution requests"""

raise NotImplementedError

def exit_work_item():

sys.exit(0)

if sys.platform == 'cli':

# We need special handling to reset the abort for keyboard interrupt exceptions

class ReplAbortException(Exception): pass

import clr

clr.AddReference('Microsoft.Dynamic')

clr.AddReference('Microsoft.Scripting')

clr.AddReference('IronPython')

from Microsoft.Scripting import KeyboardInterruptException

from Microsoft.Scripting import ParamDictionaryAttribute

from IronPython.Runtime.Operations import PythonOps

from IronPython.Runtime import PythonContext

from Microsoft.Scripting import SourceUnit, SourceCodeKind

from Microsoft.Scripting.Runtime import Scope

python_context = clr.GetCurrentRuntime().GetLanguage(PythonContext)

from System import DBNull, ParamArrayAttribute

builtin_method_descriptor_type = type(list.append)

import System

NamespaceType = type(System)

class _OldClass:

pass

_OldClassType = type(_OldClass)

_OldInstanceType = type(_OldClass())

class BasicReplBackend(ReplBackend):

future_bits = 0x3e010 # code flags used to mark future bits

"""Basic back end which executes all Python code in-proc"""

def __init__(self, mod_name='__main__'):

import threading

ReplBackend.__init__(self)

if mod_name is not None:

if sys.platform == 'cli':

self.exec_mod = Scope()

self.exec_mod.__name__ = '__main__'

else:

sys.modules[mod_name] = self.exec_mod = imp.new_module(mod_name)

else:

self.exec_mod = sys.modules['__main__']

self.code_flags = 0

self.execute_item = None

self.execute_item_lock = threading.Lock()

self.execute_item_lock.acquire() # lock starts acquired (we use it like manual reset event)

def init_connection(self):

sys.stdout = _ReplOutput(self, is_stdout = True)

sys.stderr = _ReplOutput(self, is_stdout = False)

sys.stdin = _ReplInput(self)

if sys.platform == 'cli':

import System

System.Console.SetOut(DotNetOutput(self, True))

System.Console.SetError(DotNetOutput(self, False))

def connect(self, port):

ReplBackend.connect(self, port)

self.init_connection()

def connect_using_socket(self, socket):

ReplBackend.connect_using_socket(self, socket)

self.init_connection()

def run_file_as_main(self, filename, args):

f = open(filename, 'rb')

try:

contents = f.read().replace(to_bytes('\r\n'), to_bytes('\n'))

finally:

f.close()

sys.argv = [filename]

sys.argv.extend(_command_line_to_args_list(args))

self.exec_mod.__file__ = filename

if sys.platform == 'cli':

code = python_context.CreateSnippet(contents, None, SourceCodeKind.File)

code.Execute(self.exec_mod)

else:

self.code_flags = 0

real_file = filename

if isinstance(filename, unicode) and unicode is not str:

# http://pytools.codeplex.com/workitem/696

# We need to encode the unicode filename here, Python 2.x will throw trying

# to convert it to ASCII instead of the filesystem encoding.

real_file = filename.encode(sys.getfilesystemencoding())

code = compile(contents, real_file, 'exec')

self.code_flags |= (code.co_flags & BasicReplBackend.future_bits)

exec(code, self.exec_mod.__dict__, self.exec_mod.__dict__)

def python_executor(self, code):

"""we can't close over unbound variables in execute_code_work_item

due to the exec, so we do it here"""

def func():

code.Execute(self.exec_mod)

return func

def execute_code_work_item(self):

_debug_write('Executing: ' + repr(self.current_code))

stripped_code = self.current_code.strip()

if stripped_code:

if sys.platform == 'cli':

code_to_send = ''

for line in stripped_code.split('\n'):

stripped = line.strip()

if (stripped.startswith('#') or not stripped) and not code_to_send:

continue

code_to_send += line + '\n'

code = python_context.CreateSnippet(code_to_send, None, SourceCodeKind.InteractiveCode)

dispatcher = clr.GetCurrentRuntime().GetLanguage(PythonContext).GetCommandDispatcher()

if dispatcher is not None:

dispatcher(self.python_executor(code))

else:

code.Execute(self.exec_mod)

else:

code = compile(self.current_code, '<stdin>', 'single', self.code_flags)

self.code_flags |= (code.co_flags & BasicReplBackend.future_bits)

exec(code, self.exec_mod.__dict__, self.exec_mod.__dict__)

self.current_code = None

def run_one_command(self, cur_modules, cur_ps1, cur_ps2):

# runs a single iteration of an input, execute file, etc...

# This is extracted into it's own method so we play nice w/ IronPython thread abort.

# Otherwise we have a nested exception hanging around and the 2nd abort doesn't

# work (that's probably an IronPython bug)

try:

new_modules = self._get_cur_module_set()

try:

if new_modules != cur_modules:

self.send_modules_changed()

except:

pass

cur_modules = new_modules

self.execute_item_lock.acquire()

cur_cwd = os.getcwd()

if self.check_for_exit_execution_loop():

return True, None, None, None

if self.execute_item is not None:

try:

self.execute_item()

finally:

self.execute_item = None

try:

self.send_command_executed()

except SocketError:

return True, None, None, None

try:

if cur_ps1 != sys.ps1 or cur_ps2 != sys.ps2:

new_ps1 = str(sys.ps1)

new_ps2 = str(sys.ps2)

self.send_prompt(new_ps1, new_ps2, allow_multiple_statements=False)

cur_ps1 = new_ps1

cur_ps2 = new_ps2

except Exception:

pass

try:

if cur_cwd != os.getcwd():

self.send_cwd()

except Exception:

pass

except SystemExit:

self.send_error()

self.send_exit()

# wait for ReplEvaluator to send back exit requested which will indicate

# that all the output has been processed.

while not self.exit_requested:

time.sleep(.25)

return True, None, None, None

except BaseException:

_debug_write('Exception')

exc_type, exc_value, exc_tb = sys.exc_info()

if sys.platform == 'cli':

if isinstance(exc_value.clsException, System.Threading.ThreadAbortException):

try:

System.Threading.Thread.ResetAbort()

except SystemError:

pass

sys.stderr.write('KeyboardInterrupt')

else:

# let IronPython format the exception so users can do -X:ExceptionDetail or -X:ShowClrExceptions

exc_next = self.skip_internal_frames(exc_tb)

sys.stderr.write(''.join(traceback.format_exception(exc_type, exc_value, exc_next)))

else:

exc_next = self.skip_internal_frames(exc_tb)

sys.stderr.write(''.join(traceback.format_exception(exc_type, exc_value, exc_next)))

try:

self.send_error()

except SocketError:

_debug_write('err sending DONE')

return True, None, None, None

return False, cur_modules, cur_ps1, cur_ps2

def skip_internal_frames(self, tb):

"""return the first frame outside of the repl/debugger code"""

while tb is not None and self.is_internal_frame(tb):

tb = tb.tb_next

return tb

def is_internal_frame(self, tb):

"""return true if the frame is from internal code (repl or debugger)"""

f = tb.tb_frame

co = f.f_code

filename = co.co_filename

return filename.endswith('visualstudio_py_repl.py') or filename.endswith('visualstudio_py_debugger.py')

def execution_loop(self):

"""loop on the main thread which is responsible for executing code"""

if sys.platform == 'cli' and sys.version_info[:3] < (2, 7, 1):

# IronPython doesn't support thread.interrupt_main until 2.7.1

import System

self.main_thread = System.Threading.Thread.CurrentThread

# save ourselves so global lookups continue to work (required pre-2.6)...

cur_modules = set()

try:

cur_ps1 = sys.ps1

cur_ps2 = sys.ps2

except:

# CPython/IronPython don't set sys.ps1 for non-interactive sessions, Jython and PyPy do

sys.ps1 = cur_ps1 = '>>> '

sys.ps2 = cur_ps2 = '... '

self.send_prompt(cur_ps1, cur_ps2, allow_multiple_statements=False)

while True:

exit, cur_modules, cur_ps1, cur_ps2 = self.run_one_command(cur_modules, cur_ps1, cur_ps2)

if exit:

return

def check_for_exit_execution_loop(self):

return False

def execute_script_work_item(self):

self.run_file_as_main(self.current_code, self.current_args)

def execute_module_work_item(self):

new_argv = [''] + _command_line_to_args_list(self.current_args)

old_argv = sys.argv

import runpy

try:

sys.argv = new_argv

runpy.run_module(self.current_code, alter_sys=True)

except Exception:

traceback.print_exc()

finally:

sys.argv = old_argv

def execute_process_work_item(self):

try:

from subprocess import Popen, PIPE, STDOUT

import codecs

out_codec = codecs.lookup(sys.stdout.encoding)

proc = Popen(

'"%s" %s' % (self.current_code, self.current_args),

stdout=PIPE,

stderr=STDOUT,

bufsize=0,

)

for line in proc.stdout:

print(out_codec.decode(line, 'replace')[0].rstrip('\r\n'))

except Exception:

traceback.print_exc()

@staticmethod

def _get_cur_module_set():

"""gets the set of modules avoiding exceptions if someone puts something

weird in there"""

try:

return set(sys.modules)

except:

res = set()

for name in sys.modules:

try:

res.add(name)

except:

pass

return res

def run_command(self, command):

self.current_code = command

self.execute_item = self.execute_code_work_item

self.execute_item_lock.release()

def execute_file_ex(self, filetype, filename, args):

self.current_code = filename

self.current_args = args

self.execute_item = getattr(self, 'execute_%s_work_item' % filetype, None)

self.execute_item_lock.release()

def interrupt_main(self):

# acquire the send lock so we dont interrupt while we're communicting w/ the debugger

with self.send_lock:

if sys.platform == 'cli' and sys.version_info[:3] < (2, 7, 1):

# IronPython doesn't get thread.interrupt_main until 2.7.1

self.main_thread.Abort(ReplAbortException())

else:

thread.interrupt_main()

def exit_process(self):

self.execute_item = exit_work_item

try:

self.execute_item_lock.release()

except:

pass

sys.exit(0)

def get_members(self, expression):

"""returns a tuple of the type name, instance members, and type members"""

getattr_func = getattr

if not expression:

all_members = {}

if sys.platform == 'cli':

code = python_context.CreateSnippet('vars()', None, SourceCodeKind.AutoDetect)

items = code.Execute(self.exec_mod)

else:

items = self.exec_mod.__dict__

for key, value in items.items():

all_members[key] = self.get_type_name(value)

return '', all_members, {}

else:

if sys.platform == 'cli':

code = python_context.CreateSnippet(expression, None, SourceCodeKind.AutoDetect)

val = code.Execute(self.exec_mod)

code = python_context.CreateSnippet('dir(' + expression + ')', None, SourceCodeKind.AutoDetect)

members = code.Execute(self.exec_mod)

code = python_context.CreateSnippet('lambda value, name: getattr(value, name)', None, SourceCodeKind.AutoDetect)

getattr_func = code.Execute(self.exec_mod)

else:

val = eval(expression, self.exec_mod.__dict__, self.exec_mod.__dict__)

members = dir(val)

return self.collect_members(val, members, getattr_func)

def collect_members(self, val, members, getattr_func):

t = type(val)

inst_members = {}

if hasattr(val, '__dict__'):

# collect the instance members

try:

for mem_name in val.__dict__:

mem_t = self._get_member_type(val, mem_name, True, getattr_func)

if mem_t is not None:

inst_members[mem_name] = mem_t

except:

pass

# collect the type members

type_members = {}

for mem_name in members:

if mem_name not in inst_members:

mem_t = self._get_member_type(val, mem_name, False, getattr_func)

if mem_t is not None:

type_members[mem_name] = mem_t

return t.__module__ + '.' + t.__name__, inst_members, type_members

def get_ipy_sig(self, obj, ctor):

args = []

vargs = None

varkw = None

defaults = []

for param in ctor.GetParameters():

if param.IsDefined(ParamArrayAttribute, False):

vargs = param.Name

elif param.IsDefined(ParamDictionaryAttribute, False):

varkw = param.Name

else:

args.append(param.Name)

if param.DefaultValue is not DBNull.Value:

defaults.append(repr(param.DefaultValue))

return obj.__doc__, args, vargs, varkw, tuple(defaults)

def get_signatures(self, expression):

if sys.platform == 'cli':

code = python_context.CreateSnippet(expression, None, SourceCodeKind.AutoDetect)

val = code.Execute(self.exec_mod)

else:

val = eval(expression, self.exec_mod.__dict__, self.exec_mod.__dict__)

return self.collect_signatures(val)

def collect_signatures(self, val):

doc = val.__doc__

type_obj = None

if isinstance(val, type) or isinstance(val, _OldClassType):

type_obj = val

val = val.__init__

try:

args, vargs, varkw, defaults = inspect.getargspec(val)

except TypeError:

# we're not doing inspect on a Python function...

if sys.platform == 'cli':

if type_obj is not None:

clr_type = clr.GetClrType(type_obj)

ctors = clr_type.GetConstructors()

return [self.get_ipy_sig(type_obj, ctor) for ctor in ctors]

elif type(val) is types.BuiltinFunctionType:

return [self.get_ipy_sig(target, target.Targets[0]) for target in val.Overloads.Functions]

elif type(val) is builtin_method_descriptor_type:

val = PythonOps.GetBuiltinMethodDescriptorTemplate(val)

return [self.get_ipy_sig(target, target.Targets[0]) for target in val.Overloads.Functions]

raise

remove_self = type_obj is not None or (type(val) is types.MethodType and

((sys.version_info >= (3,) and val.__self__ is not None) or

(sys.version_info < (3,) and val.im_self is not None)))

if remove_self:

# remove self for instance methods and types

args = args[1:]

if defaults is not None:

defaults = [repr(default) for default in defaults]

else:

defaults = []

return [(doc, args, vargs, varkw, defaults)]

def set_current_module(self, module):

mod = sys.modules.get(module)

if mod is not None:

_debug_write('Setting module to ' + module)

if sys.platform == 'cli':

self.exec_mod = clr.GetClrType(type(sys)).GetProperty('Scope', System.Reflection.BindingFlags.Public | System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance).GetValue(sys, ())

else:

self.exec_mod = mod

elif module:

_debug_write('Unknown module ' + module)

def get_module_names(self):

res = []

for name, module in sys.modules.items():

try:

if name != 'visualstudio_py_repl' and name != '$visualstudio_py_debugger':

if sys.platform == 'cli' and type(module) is NamespaceType:

self.get_namespaces(name, module, res)

else:

try:

filename = os.path.abspath(module.__file__)

except Exception:

filename = None

res.append((name, filename))

except:

pass

return res

def get_namespaces(self, basename, namespace, names):

names.append((basename, ''))

try:

for name in dir(namespace):

new_name = basename + '.' + name

new_namespace = getattr(namespace, name)

if type(new_namespace) is NamespaceType:

self.get_namespaces(new_name, new_namespace, names)

View remainder of file in raw view

vscode-python/pythonFiles/PythonTools/visualstudio_py_repl.py at master · hlinet/vscode-python · GitHub

Expand file tree

Breadcrumbs

visualstudio_py_repl.py

Latest commit

History

Breadcrumbs

visualstudio_py_repl.py

File metadata and controls