onyx_online/obsws-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/obsws-python.git synced 2024-11-25 06:10:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

add values to defautlkwargs.

Browse Source 
if no conn kwargs then check and merge config.toml

raise error if auth enabled but no password

add full request payload to logger
This commit is contained in:
onyx-and-iris 2022-11-17 11:30:39 +00:00
parent c4ee2347ba
commit 9fa5b3f4b4
1 changed files with 19 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
obsws_python/baseclient.py
View File

@ -19,28 +19,30 @@ class ObsClient:
logger = logging.getLogger("baseclient.obsclient") logger = logging.getLogger("baseclient.obsclient")
def __init__(self, **kwargs): def __init__(self, **kwargs):
defaultkwargs = { defaultkwargs = {"host": "localhost", "port": 4455, "password": None, "subs": 0}
**{key: None for key in ["host", "port", "password"]}, if not any(key in kwargs for key in ("host", "port", "password")):
"subs": 0, kwargs |= self._conn_from_toml()
}
kwargs = defaultkwargs | kwargs kwargs = defaultkwargs | kwargs
for attr, val in kwargs.items(): for attr, val in kwargs.items():
setattr(self, attr, val) setattr(self, attr, val)
if not (self.host and self.port):
conn = self._conn_from_toml() self.logger.info(
self.host = conn["host"] "Connecting with parameters: {host} {port} {password} {subs}".format(
self.port = conn["port"] **self.__dict__
self.password = conn["password"] )
)
self.ws = websocket.WebSocket() self.ws = websocket.WebSocket()
self.ws.connect(f"ws://{self.host}:{self.port}") self.ws.connect(f"ws://{self.host}:{self.port}")
self.server_hello = json.loads(self.ws.recv()) self.server_hello = json.loads(self.ws.recv())
def _conn_from_toml(self): def _conn_from_toml(self) -> dict:
conn = {}
filepath = Path.cwd() / "config.toml" filepath = Path.cwd() / "config.toml"
if filepath.exists():
with open(filepath, "rb") as f: with open(filepath, "rb") as f:
conn = tomllib.load(f) conn |= tomllib.load(f)
return conn["connection"] return conn["connection"] if "connection" in conn else conn
def authenticate(self): def authenticate(self):
payload = { payload = {
@ -52,6 +54,8 @@ class ObsClient:
} }
if "authentication" in self.server_hello["d"]: if "authentication" in self.server_hello["d"]:
if not self.password:
raise OBSSDKError("authentication enabled but no password provided")
secret = base64.b64encode( secret = base64.b64encode(
hashlib.sha256( hashlib.sha256(
( (
@ -79,14 +83,13 @@ class ObsClient:
raise OBSSDKError("failed to identify client with the server") raise OBSSDKError("failed to identify client with the server")
def req(self, req_type, req_data=None): def req(self, req_type, req_data=None):
id = randint(1, 1000)
self.logger.debug(f"Sending request with id {id}")
payload = { payload = {
"op": 6, "op": 6,
"d": {"requestType": req_type, "requestId": id}, "d": {"requestType": req_type, "requestId": randint(1, 1000)},
} }
if req_data: if req_data:
payload["d"]["requestData"] = req_data payload["d"]["requestData"] = req_data
self.logger.debug(f"Sending request {payload}")
self.ws.send(json.dumps(payload)) self.ws.send(json.dumps(payload))
response = json.loads(self.ws.recv()) response = json.loads(self.ws.recv())
self.logger.debug(f"Response received {response}") self.logger.debug(f"Response received {response}")