Source.Python/docs/_modules/core.html at master · Source-Python-Dev-Team/Source.Python · GitHub

Repository navigation

History

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

<!DOCTYPE html>

<head>

<title>core — Source.Python v743 documentation</title>

</head><body>

<h3>Navigation</h3>

<ul>

<a href="../genindex.html" title="General Index"

accesskey="I">index</a></li>

<a href="../py-modindex.html" title="Python Module Index"

>modules</a> |</li>

<li class="nav-item nav-item-0"><a href="../index.html"> Source.Python v743 documentation</a> »</li>

<li class="nav-item nav-item-1"><a href="index.html" accesskey="U">Module code</a> »</li>

</ul>

</div>

<h1>Source code for core</h1><div class="highlight"><pre>

# ../core/__init__.py

"""Provides core functionality that doesn't fit into any other package."""

# =============================================================================

# >> IMPORTS

# =============================================================================

# Python Imports

# Codecs

import codecs

# Collections

from collections import defaultdict

# Contextlib

from contextlib import contextmanager

# Hashlib

import hashlib

# Inspect

from inspect import getmodule

from inspect import getmodulename

from inspect import currentframe

# OS

from os import sep

# Path

from path import Path

# Platform

from platform import system

# RE

from re import compile as re_compile

from re import finditer

# Sys

import sys

# Urllib

from urllib.request import urlopen

# Weakref

from weakref import WeakValueDictionary

# Site-Packages Imports

# ConfigObj

from configobj import ConfigObj

# Source.Python Imports

# Paths

from paths import GAME_PATH

from paths import CFG_PATH

from paths import PLUGIN_PATH

# =============================================================================

# >> FORWARD IMPORTS

# =============================================================================

# Source.Python Imports

# Core

from _core import BoostPythonClass

from _core import console_message

from _core import get_interface

from _core import get_core_modules

from _core import OutputReturn

from _core import SOURCE_ENGINE

from _core import SOURCE_ENGINE_BRANCH

# =============================================================================

# >> ALL DECLARATION

# =============================================================================

__all__ = ('AutoUnload',

'BoostPythonClass',

'ConfigFile',

'GameConfigObj',

'WeakAutoUnload',

'GAME_NAME',

'OutputReturn',

'PLATFORM',

'SOURCE_ENGINE',

'SOURCE_ENGINE_BRANCH',

'Tokenize',

'check_info_output',

'console_message',

'create_checksum',

'echo_console',

'get_core_modules',

'get_interface',

'get_public_ip',

'ignore_unicode_errors',

'server_output',

)

# =============================================================================

# >> GLOBAL VARIABLES

# =============================================================================

# Get the specific game for the server

GAME_NAME = GAME_PATH.name

# Get the platform the server is on

PLATFORM = system().lower()

# =============================================================================

# >> CLASSES

# =============================================================================

class AutoUnload:

"""Class used to auto unload specific instances.

Each inheriting class must implement an _unload_instance method.

"""

# Create a dictionary to store AutoUnload objects in

_module_instances = defaultdict(list)

def __new__(cls, *args, **kwargs):

"""Overwrite __new__ to store the calling module."""

# Get the class instance

self = super().__new__(cls)

# Get the calling frame

frame = currentframe().f_back

# Get the calling path

path = frame.f_code.co_filename

# Don't keep hostage instances that will never be unloaded

while not path.startswith(PLUGIN_PATH):

frame = frame.f_back

if frame is None:

return self

path = frame.f_code.co_filename

if path.startswith('<frozen'):

return self

# Resolve the calling module name

try:

name = frame.f_globals['__name__']

except KeyError:

try:

name = getmodule(frame).__name__

except AttributeError:

name = getmodulename(path)

# Call class-specific logic for adding the instance.

if name is not None:

self._add_instance(name)

# Return the instance

return self

def _add_instance(self, caller):

"""Add the instance to self._module_instances."""

self._module_instances[caller].append(self)

def _unload_instance(self):

"""Base _unload_instance implementation."""</div>

class WeakAutoUnload(AutoUnload):

"""Subclass of AutoUnload used to store weak references to instances."""

# Create a dictionary to store WeakAutoUnload objects in

_module_instances = defaultdict(WeakValueDictionary)

"""Add the instance to self._module_instances."""

self._module_instances[caller][id(self)] = self</div>

class GameConfigObj(ConfigObj):

"""Class used to parse specific game data."""

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

"""Helper class that merges the given file with engine/game files."""

# Get the file directory/name...

path, name = Path(infile).splitpath()

# Call ConfigObj's __init__ method...

super().__init__(infile, *args, **kwargs)

# Move the path to the current engine sub-directory...

path = path / SOURCE_ENGINE

# Parse and merge the specific engine file...

self.merge(ConfigObj(path / name, *args, **kwargs))

# Finally, parse the specific game file...

self.merge(ConfigObj(path / GAME_NAME / name, *args, **kwargs))</div>

</div>

class Tokenize(list):

"""Parses the arguments from the given string."""

_pattern = re_compile(r'"[^"]*"|[^ \t]+')

def __init__(self, string, comment_prefix=None):

"""Splits the arguments from the given string."""

# Initialize the list

super().__init__()

# Store the given string as is

self.string = string

# Loop through all tokens

for match in finditer(self._pattern, string):

# Get the current match as a string

arg = match.group()

# Strip end line comment

if comment_prefix is not None and arg.startswith(comment_prefix):

self.string = self.string[:match.start()]

break

# Add the current argument to the list

self.append(arg.strip('"'))</div>

def __str__(self):

"""Returns the original string (without end-line comment)."""

return self.string

def __hash__(self):

"""Hashes the original string."""

return hash(self.string)</div>

class ConfigFile(list):

"""Class used to parse a configuration file."""

def __init__(

self, path, encoding='utf-8', comment_prefix='//', as_strings=False):

"""Parses the given configuation file path.

:param Path path:

The path of the file to parse.

:param str encoding:

The encoding to use when opening the file.

:param str comment_prefix:

The prefix of end line comments.

:param bool as_strings:

Whether the parsed lines should be stored as strings rather than

argument lists.

"""

# If the given path doesn't exist, search for it in the cfg directory

if not path.is_file():

path = CFG_PATH.joinpath(path)

# If no file was found, return an empty list

if not path.is_file():

return

# Import this here to fix cyclic imports

from translations.strings import LangStrings

# Open the given file and parse its content

with open(path, 'r', encoding=encoding) as f:

# Loop through all lines

for line in f.read().splitlines():

# Parse the argument from the current line

args = Tokenize(

LangStrings._replace_escaped_sequences(line),

comment_prefix)

# Skip empty/commented lines

if not args:

continue

# Add the current line to the list

self.append(args if not as_strings else str(args))</div>

</div>

# =============================================================================

# >> FUNCTIONS

# =============================================================================

def echo_console(text):

"""Echo a message to the server's console.

.. note::

Unlike ``console_message``, this function automatically adds a newline

at the end of the message.

:param str text:

Message to print to the console.

"""

console_message(text + '\n')</div>

@contextmanager

def ignore_unicode_errors(errors='ignore'):

"""Overwrite the ``strict`` codecs error handler temporarily.

This is useful e.g. if the engine truncates a string, which results in a

string that contains a splitted multi-byte character at the end of the

string.

:param str errors:

Error handler that will be looked up via :func:`codecs.lookup_error`.

:raise LookupError:

Raised if the error handler was not found.

Example:

.. code:: python

import memory

# Allocate four bytes to create an erroneous string

ptr = memory.alloc(4)

# Write data to the memory that will usually result in a

# UnicodeDecodeError

ptr.set_uchar(ord('a'), 0)

ptr.set_uchar(ord('b'), 1)

ptr.set_uchar(226, 2) # Add the invalid byte

ptr.set_uchar(0, 3) # Indicate the end of the string

with ignore_unicode_errors():

# Read the data as a string. Now, it will only print 'ab', because

# the invalid byte has been removed/ignored.

print(ptr.get_string_array())

"""

old_handler = codecs.lookup_error('strict')

codecs.register_error('strict', codecs.lookup_error(errors))

try:

yield

finally:

def get_public_ip():

"""Return the server's public IPv4.

:rtype: str

.. note::

This functions makes a call to ``http://api.ipify.org`` to retrieve the public IP.

"""

return urlopen('http://api.ipify.org/').read().decode()</div>

@contextmanager

def server_output(action=OutputReturn.CONTINUE):

"""Gather all server output sent during the execution of the with-statement.

:param OutputReturn action:

Determines what happens with the output.

:rtype: list

:return:

A list that will be filled with a tuple for every line that is being

logged. The tuple contains the severity and the message.

Example:

.. code:: python

from cvars import cvar

from core import server_output

from core import OutputReturn

status = cvar.find_command('status')

with server_output(OutputReturn.BLOCK) as output:

status()

# Print everything that was logged by the 'status' command

print(output)

Output:

.. code:: python

[(_core.MessageSeverity.MESSAGE, 'hostname: Counter-Strike: Global Offensive\\n'),

(_core.MessageSeverity.MESSAGE, 'version : 1.35.8.4/13584 513/6771 secure [A:1:2435270659:8640] \\n'),

(_core.MessageSeverity.MESSAGE, 'udp/ip : 192.168.178.60:27015 (public ip: 46.83.158.27)\\n'),

(_core.MessageSeverity.MESSAGE, 'os : Windows\\n'),

(_core.MessageSeverity.MESSAGE, 'type : community dedicated\\n'),

(_core.MessageSeverity.MESSAGE, 'players : 0 humans, 0 bots (20/0 max) (hibernating)\\n\\n'),

(_core.MessageSeverity.MESSAGE, '# userid name uniqueid connected ping loss state rate'),

(_core.MessageSeverity.MESSAGE, ' adr'),

(_core.MessageSeverity.MESSAGE, '\\n'),

(_core.MessageSeverity.MESSAGE, '#end\\n')]

"""

# Import this here to fix a cyclic import

from listeners import OnServerOutput

msg_buffer = []

def intercepter(severity, msg):

msg_buffer.append((severity, msg))

return action

OnServerOutput.manager.register_listener(intercepter)

try:

yield msg_buffer

finally:

OnServerOutput.manager.unregister_listener(intercepter)</div>

def create_checksum(data, ignore_wchars=True):

"""Create an MD5 checksum for the given string.

:param str data:

The string for which a checksum should be created.

:param bool ignore_wchars:

If ``True`` whitespace characters are ignored.

:rtype: str

"""

if ignore_wchars:

data = ''.join(data.split())

return hashlib.new('md5', bytes(data, encoding='utf-8')).hexdigest()</div>

def check_info_output(output):

"""Return whether the output of ``sp info`` has been modified.

:param str output:

The output of ``sp info``.

:raise ValueError:

Raised if the checksum was not found in the output.

:return:

``True`` if the output has been modified.

:rtype: bool

"""

checksum = None

lines = output.strip().split('\n')

# Search the checksum entry

while lines:

line = lines.pop(0)

if line.startswith('Checksum'):

checksum = line.split(':', 1)[1].strip()

break

if checksum is None:

raise ValueError('Checksum not found.')

# Ignore last line if it's the separator

if lines[-1].startswith('-'):

lines.pop()

return create_checksum(''.join(lines)) != checksum</div>

</pre></div>

</div>

</a>

<h3 id="searchlabel">Quick search</h3>

</form>

</div>

</search>

</div>

</div>

<h3>Navigation</h3>

<ul>

<a href="../genindex.html" title="General Index"

>index</a></li>

<a href="../py-modindex.html" title="Python Module Index"

>modules</a> |</li>

<li class="nav-item nav-item-0"><a href="../index.html"> Source.Python v743 documentation</a> »</li>

<li class="nav-item nav-item-1"><a href="index.html" >Module code</a> »</li>

</ul>

</div>

Created using <a href="https://www.sphinx-doc.org/">Sphinx</a> 8.2.3.

</div>

</body>

</html>

Source.Python/docs/_modules/core.html at master · Source-Python-Dev-Team/Source.Python · GitHub

Repository navigation

Expand file tree

Breadcrumbs

core.html

Latest commit

History

Breadcrumbs

core.html

File metadata and controls