65 lines
1.6 KiB
Python
65 lines
1.6 KiB
Python
from typing import Dict, List, Union
|
|
from logging import Logger
|
|
import json
|
|
|
|
if False:
|
|
from .connection import Connection
|
|
from .simplenode import SimpleNode
|
|
|
|
|
|
LOGGER = Logger(__name__)
|
|
|
|
|
|
def _build_patch_set(change: dict) -> str:
|
|
line = ['^', change['path'], ' : ']
|
|
if change['type'] in ('Long', 'Float', 'Boolean', 'String'):
|
|
line.append(json.dumps(change['value']))
|
|
else:
|
|
raise ValueError(f'Type {change["type"]!r} is currently not supported!')
|
|
|
|
return ''.join(line)
|
|
|
|
|
|
_patch_builders = {
|
|
'set': _build_patch_set
|
|
}
|
|
|
|
|
|
class PatchBuilder:
|
|
changes: List[Dict]
|
|
|
|
def __init__(self, connection: 'Connection'):
|
|
self.connection = connection
|
|
self.changes = []
|
|
self.saved = False
|
|
|
|
def set_value(self, path: str, value: Union[str, int, float], value_type: str):
|
|
self.changes.append({
|
|
'action': 'set',
|
|
'path': path,
|
|
'value': value,
|
|
'type': value_type
|
|
})
|
|
|
|
def save(self):
|
|
patch = []
|
|
for entry in self.changes:
|
|
patch.append(_patch_builders[entry['action']](entry))
|
|
patch = list(filter(None, patch))
|
|
if len(patch) == 0:
|
|
LOGGER.warning("No patch to submit")
|
|
return
|
|
|
|
self.connection.apply_diff('\n'.join(patch))
|
|
self.saved = True
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if not self.saved:
|
|
LOGGER.warning("Patch not saved")
|
|
|
|
if self.connection:
|
|
self.connection._patch_builder = None
|