1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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
|
"""
Part of the tagit module.
A copy of the license is provided with the project.
Author: Matthias Baumgartner, 2022
"""
# standard imports
import os
import typing
# kivy imports
from kivy.clock import Clock
from kivy.uix.widget import Widget
import kivy.properties as kp
# tagit imports
from tagit import parsing
from tagit.config.loader import load_settings
from tagit.utils import bsfs
# exports
__all__: typing.Sequence[str] = (
'ConfigAwareMixin',
'Session',
)
## code ##
class Session(Widget):
storage = kp.ObjectProperty(None)
cfg = kp.ObjectProperty(None)
__events__ = ('on_storage_modified', 'on_predicate_modified', 'on_config_changed')
def __init__(self, cfg, storage, log, **kwargs):
super(Session, self).__init__(**kwargs)
self.cfg = cfg
self.storage = storage
self.log = log
# derived members
self.filter_from_string = parsing.filter.FromString(self.storage.schema)
self.filter_to_string = parsing.filter.ToString(self.storage.schema)
#self.sort_from_string = parsing.Sort(self.storage.schema) # FIXME: mb/port/parsing
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
pass
def clone(self, cfg):
"""Clone the session and load the clone."""
# clone storages to new location
liburi = cfg('session', 'paths', 'library')
numuri = cfg('session', 'paths', 'numerical')
storage = Broker.Clone(self.storage, liburi, numuri, None, cfg)
log = load_log(cfg) # not cloned
# switch to new storage
self.cfg = cfg
self.log = log
self.storage = storage
def load(self, cfg):
"""Load the session from configuration *cfg*."""
#self.log = load_log(cfg) # FIXME: mb/port
# initialize storages from config
# open BSFS storage
store = bsfs.Open(cfg('session', 'bsfs'))
# check storage schema
with open(resource_find('required_schema.nt'), 'rt') as ifile:
required_schema = bsfs.schema.from_string(ifile.read())
if not required_schema.consistent_with(store.schema):
raise Exception("The storage's schema is incompatible with tagit's requirements")
if not required_schema <= store.schema:
store.migrate(required_schema | store.schema)
# replace current with new storage
self.storage = store
def update_settings_key(self, key, value):
# change setting
self.cfg.set(key, value)
# update settings file
# FIXME: file_connected is also true if it loaded config from user home!
if self.cfg.file_connected() and self.cfg('storage', 'config', 'write_through'):
# store only difference to baseline (i.e. session config)
local_config = self.cfg.diff(load_settings())
local_config.save()
# trigger update event
self.dispatch('on_config_changed', key, value)
def on_config_changed(sender, key, value):
"""Event prototype."""
pass
def on_storage(self, wx, storage):
# fire event if the storage was replaced
self.dispatch('on_storage_modified')
def on_storage_modified(sender):
"""Event prototype.
Triggered when items are added or removed
"""
pass
def on_predicate_modified(sender, predicate, objects, diff):
"""Event prototype.
Triggered when a predicate to one or several objects have been changed.
"""
pass
class StorageAwareMixin(object):
def on_root(self, wx, root):
session = root.session
# storage has been changed as a whole
session.bind(storage=self.on_storage)
# some parts of the storage have changed
session.bind(on_storage_modified=self.on_storage_modified)
session.bind(on_predicate_modified=self.on_predicate_modified)
if session.storage is not None:
# initialize with the current storage
# Going through the event dispatcher ensures that the object
# is initialized properly before on_storage is called.
Clock.schedule_once(lambda dt: self.on_storage(session, session.storage))
def on_storage(self, sender, storage):
"""Default event handler."""
pass
def on_storage_modified(self, sender):
"""Default event handler."""
pass
def on_predicate_modified(self, sender, predicate, objects, diff):
"""Default event handler."""
pass
class ConfigAwareMixin(object):
def on_root(self, wx, root):
session = root.session
# config changes as a whole
session.bind(cfg=self.on_cfg)
# individual config entries have been changed
session.bind(on_config_changed=self.on_config_changed)
if session.cfg is not None:
# initialize with the current config
# Going through the event dispatcher ensures that the object
# is initialized properly before on_cfg is called.
Clock.schedule_once(lambda dt: self.on_cfg(session, session.cfg))
def on_config_changed(self, sender, key, value):
"""Default event handler."""
pass
def on_cfg(self, sender, cfg):
"""Default event handler."""
pass
## EOF ##
|
