PK!׬^88lcu_connectorpy/__init__.pyfrom .connect import Connector __all__ = ['Connector'] PK!Qlcu_connectorpy/connect.pyimport psutil import time import yaml from pathlib import Path from watchdog import events from watchdog.observers import Observer from typing import Tuple, Callable, Optional class RestartRequiredError(Exception): def __init__(self): super().__init__("Please restart your League of Legends Client") class Lock: """Parses the data in the lock file and formats the result""" def __init__(self, path: Path): self.path = path self.data = { 'name': None, 'id': None, 'port': None, 'password': None, 'protocol': None } self.load() def load(self) -> dict: """Load the lock file, store the results, and return the stored data""" if self.path.exists(): data = self.path.read_text().split(':') self.data = {k: v for k, v in zip(self.data, data)} else: self.data = {k: None for k in self.data} return self.data @property def ready(self) -> bool: return all(self.data.values()) class LeagueClient: """Handles locating the executable and lock file. Platform independent.""" name = "LeagueClient." + 'exe' if psutil.WINDOWS else 'app' def __init__(self): self.reset() def __get_process(self) -> psutil.Process: for p in psutil.process_iter(): if self.name == p.name(): return p def __get_lock(self) -> Lock: if self.process is None: return try: files = self.process.open_files() except psutil.AccessDenied as e: self.__enable_swagger() raise RestartRequiredError from e for file in files: if file.path.endswith('lockfile'): return Lock(Path(file.path)) def __enable_swagger(self): yaml_fp = Path(self.process.exe()).with_name('system.yaml') yl = yaml.load(yaml_fp.read_text()) yl['enable_swagger'] = True yaml_fp.write_text(yaml.dump(yl)) def reset(self): """Look for the process and lock file again""" self.process = self.__get_process() self.lock = self.__get_lock() def wait(self): """Block until a process and lock file have been found""" while not self.ready: self.reset() time.sleep(1) @property def ready(self) -> bool: return bool(self.process and self.lock and self.lock.ready) class FileSentry(events.FileSystemEventHandler): def __init__(self, path: Path, callback: Callable): super().__init__() self.path = path self.callback = callback self.observer = Observer() self.observer.schedule(self, self.path.parent) def __del__(self): self.observer.stop() def on_any_event(self, event): if Path(event.src_path) == self.path: self.callback() def start(self): self.observer.start() class Connector: """ Manager for getting info required to connect to the League Client. example:: ``` import requests from lcu_connectorpy import Connector conn = Connector() conn.start() r = requests.get( f'{conn.url}/Help', auth=conn.auth, headers={'Accept': 'application/json'} ) print(r.json()) ``` """ address = '127.0.0.1' username = 'riot' port: Optional[str] = None password: Optional[str] = None protocol: Optional[str] = None id: Optional[str] = None def __init__(self): self.client = LeagueClient() self.sentry: Optional[FileSentry] = None def update(self): if not self.client.lock: return for k, v in self.client.lock.load().items(): setattr(self, k, v) def start(self): """Start watching for the client. Blocks until found.""" self.client.wait() self.update() self.sentry = FileSentry(self.client.lock.path, self.update) @property def connected(self) -> bool: """Established connection, data is ready.""" return self.client.ready @property def url(self) -> str: """A url using the current lock file data""" return f'{self.protocol}://{self.address}:{self.port}' @property def auth(self) -> Tuple[str]: """A tuple (user, password) with the latest credentials""" return (self.username, self.password) PK!H-ZZ#0lcu_connectorpy-1.1.0.dist-info/entry_points.txtN+I/N.,()JOrl..PK!*͚88'lcu_connectorpy-1.1.0.dist-info/LICENSE The MIT License (MIT) Copyright (c) 2019 Noah Corona Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HڽTU%lcu_connectorpy-1.1.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!Hk,)s(lcu_connectorpy-1.1.0.dist-info/METADATATR0}WhɌĤL3@Rnv@[`]N-PoYݳ:Syx b"r%ĴH0JA5yD,brZJmӡBySjK}t4)(O#ZBhxۙyy-?vwژeʱH@99/PʹM2iX@*u7"\#H|i (b9V+ϸ/N*c~^Sڒ cz|:Cc im#W!3˥*c$Q zn_8SJ/ 7Y/N \B!3 S 4W2(5ʾKLFWC0/_zJ/}.ܯ;4&WբA af<%Bpj v93.($"G_[Z-WI\tEH}KϞfkbz4͹E@' PS#j0<=b(v۝5Vhh`twN~+6(6Z"Ne=濑ԸPK!HlcR&lcu_connectorpy-1.1.0.dist-info/RECORDKS0໿% Xb)%hB! ^㥷=}3[g#Z΋L}7?b8 J4s%,`웒?y\fKEKȒx8*ۘp]ZE^t|:D. Be'%UVeE $࢟q2.YLVPJ}ҽMv zN_UhF8v!wk|b1*a~:Zc=qjzsk@U,XAKjjo8f[V&nJf{Ҧ3o`N2͑~F&`,Ԅ)M2l|x>|uf 0rxPK!׬^88lcu_connectorpy/__init__.pyPK!Qqlcu_connectorpy/connect.pyPK!H-ZZ#01lcu_connectorpy-1.1.0.dist-info/entry_points.txtPK!*͚88'lcu_connectorpy-1.1.0.dist-info/LICENSEPK!HڽTU%lcu_connectorpy-1.1.0.dist-info/WHEELPK!Hk,)s(lcu_connectorpy-1.1.0.dist-info/METADATAPK!HlcR&jlcu_connectorpy-1.1.0.dist-info/RECORDPKA