Skip to content

wdgrid API Documentation

pareto

Created on 2022-03-21

@author: wf

Pareto

Bases: object

Pareto level holder

Source code in wd/pareto.py 
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
class Pareto(object):
    """
    Pareto level holder
    """

    def __init__(self, level: int = 1):
        """
        Constructor

        Args:
            level(int): the pareto level

        """
        self.level = level
        self.good = 80.0
        self.bad = 20.0
        for _i in range(level - 1):
            self.good = self.good + self.bad * 0.8
            self.bad = 100.0 - self.good
        self.decimals = 1 - round(math.log10(self.bad))
        self.oneOutOf = round(100 / self.bad)
        pass

    def ratioInLevel(self, ratio) -> bool:
        """
        check whether the given ratio is in this level
        """
        inLevel = ratio >= 1 / self.oneOutOf
        return inLevel

    def asPercent(self):
        percent = 100.0 / self.oneOutOf
        return percent

    def __str__(self):
        text = self.asText(long=False)
        return text

    def asText(self, long: bool = False):
        text = f"{self.good:.{self.decimals}f}:{self.bad:.{self.decimals}f}"
        if long:
            text = f"level {self.level}={text} (1 out of {self.oneOutOf})"
        return text

    def asDict(self) -> dict:
        """
        return me as a dict

        Returns:
            dict: my values as a dict
        """
        d = {}
        d["level"] = self.level
        d["ratio"] = self.asText()
        d["1 out of"] = self.oneOutOf
        return d

__init__(level=1)

Constructor

Parameters:

Name Type Description Default
level(int)

the pareto level

 required
Source code in wd/pareto.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, level: int = 1):
    """
    Constructor

    Args:
        level(int): the pareto level

    """
    self.level = level
    self.good = 80.0
    self.bad = 20.0
    for _i in range(level - 1):
        self.good = self.good + self.bad * 0.8
        self.bad = 100.0 - self.good
    self.decimals = 1 - round(math.log10(self.bad))
    self.oneOutOf = round(100 / self.bad)
    pass

asDict()

return me as a dict

Returns:

Name Type Description
dict dict

my values as a dict
Source code in wd/pareto.py 
54
55
56
57
58
59
60
61
62
63
64
65
def asDict(self) -> dict:
    """
    return me as a dict

    Returns:
        dict: my values as a dict
    """
    d = {}
    d["level"] = self.level
    d["ratio"] = self.asText()
    d["1 out of"] = self.oneOutOf
    return d

ratioInLevel(ratio)

check whether the given ratio is in this level

Source code in wd/pareto.py 
33
34
35
36
37
38
def ratioInLevel(self, ratio) -> bool:
    """
    check whether the given ratio is in this level
    """
    inLevel = ratio >= 1 / self.oneOutOf
    return inLevel

query_view

Created on 2024-01-04

@author: wf

QueryView

widget to display queries

Source code in wd/query_view.py 
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
class QueryView:
    """
    widget to display queries
    """

    def __init__(
        self, webserver: NiceGuiWebserver, name: str, sparql_endpoint: Endpoint
    ):
        """
        Initialize the QueryView object with a given webserver and name.

        Args:
            webserver (NiceGuiWebserver): The web server instance to be used.
            name (str): The name identifier for the query display.
            sparql_endpoint(endpoint): the SPARQL endpoint to use
        """
        self.webserver = webserver
        self.name = name
        self.setup()
        self.sparql_query = ""
        self.sparql_markup = ""
        self.sparql_endpoint = sparql_endpoint

    def setup(self):
        """Set up the UI components for the query display."""
        with ui.expansion(self.name) as self.expansion:
            self.code_view = ui.code("", language="sparql")
        with ui.row() as self.link_row:
            self.try_it_link_view = ui.html()
            self.download_link_view = ui.html()
            pass

    def show_query(self, sparql_query: str):
        """
        Update the display with a new SPARQL query.

        Args:
            sparql_query (str): The SPARQL query string to be displayed.
        """
        self.sparql_query = sparql_query.strip()
        # we might need to change the endpoint
        self.query = Query(name=self.name, query=sparql_query)
        if self.sparql_endpoint:
            try_it_url_encoded = self.query.getTryItUrl(
                baseurl=self.sparql_endpoint.website,
                database=self.sparql_endpoint.database,
            )
            with self.link_row:
                try_it_link = Link.create(try_it_url_encoded, "try it!")
                self.try_it_link_view.content = try_it_link
        with self.expansion:
            self.code_view.markdown.content = f"""```sparql
    {sparql_query}
    ```"""
            self.code_view.update()
        pass

__init__(webserver, name, sparql_endpoint)

Initialize the QueryView object with a given webserver and name.

Parameters:

Name Type Description Default
webserver NiceGuiWebserver

The web server instance to be used.

 required
name str

The name identifier for the query display.

 required
sparql_endpoint(endpoint)

the SPARQL endpoint to use

 required
Source code in wd/query_view.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def __init__(
    self, webserver: NiceGuiWebserver, name: str, sparql_endpoint: Endpoint
):
    """
    Initialize the QueryView object with a given webserver and name.

    Args:
        webserver (NiceGuiWebserver): The web server instance to be used.
        name (str): The name identifier for the query display.
        sparql_endpoint(endpoint): the SPARQL endpoint to use
    """
    self.webserver = webserver
    self.name = name
    self.setup()
    self.sparql_query = ""
    self.sparql_markup = ""
    self.sparql_endpoint = sparql_endpoint

setup()

Set up the UI components for the query display.

Source code in wd/query_view.py 
36
37
38
39
40
41
42
43
def setup(self):
    """Set up the UI components for the query display."""
    with ui.expansion(self.name) as self.expansion:
        self.code_view = ui.code("", language="sparql")
    with ui.row() as self.link_row:
        self.try_it_link_view = ui.html()
        self.download_link_view = ui.html()
        pass

show_query(sparql_query)

Update the display with a new SPARQL query.

Parameters:

Name Type Description Default
sparql_query str

The SPARQL query string to be displayed.

 required
Source code in wd/query_view.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def show_query(self, sparql_query: str):
    """
    Update the display with a new SPARQL query.

    Args:
        sparql_query (str): The SPARQL query string to be displayed.
    """
    self.sparql_query = sparql_query.strip()
    # we might need to change the endpoint
    self.query = Query(name=self.name, query=sparql_query)
    if self.sparql_endpoint:
        try_it_url_encoded = self.query.getTryItUrl(
            baseurl=self.sparql_endpoint.website,
            database=self.sparql_endpoint.database,
        )
        with self.link_row:
            try_it_link = Link.create(try_it_url_encoded, "try it!")
            self.try_it_link_view.content = try_it_link
    with self.expansion:
        self.code_view.markdown.content = f"""```sparql
{sparql_query}
```"""
        self.code_view.update()
    pass

truly_tabular_display

Created on 2024-01-04

@author: wf

PropertySelection

select properties

Source code in wd/truly_tabular_display.py 
 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
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
class PropertySelection:
    """
    select properties
    """

    def __init__(
        self,
        inputList,
        total: int,
        paretoLevels: Dict[int, Pareto],
        minFrequency: float,
    ):
        """
           Constructor

        Args:
            propertyList(list): the list of properties to show
            total(int): total number of properties
            paretolLevels: a dict of paretoLevels with the key corresponding to the level
            minFrequency(float): the minimum frequency of the properties to select in percent
        """
        self.propertyMap: Dict[str, dict] = dict()
        self.headerMap = {}
        self.propertyList = []
        self.total = total
        self.paretoLevels = paretoLevels
        self.minFrequency = minFrequency
        for record in inputList:
            ratio = int(record["count"]) / self.total
            level = self.getParetoLevel(ratio)
            record["%"] = f"{ratio*100:.1f}"
            record["pareto"] = level
            # if record["pareto"]<=paretoLimit:
            orecord = collections.OrderedDict(record.copy())
            self.propertyList.append(orecord)
        pass

    @property
    def aggregates(self) -> list:
        aggregates = ["min", "max", "avg", "sample", "list", "count"]
        return aggregates

    @property
    def option_cols(self) -> list:
        option_cols = ["ignore", "label"]
        return option_cols

    @property
    def checkbox_cols(self) -> list:
        """
        get all my checkbox columns
        """
        checkbox_cols = self.aggregates
        checkbox_cols.extend(self.option_cols)
        return checkbox_cols

    def getParetoLevel(self, ratio):
        level = 0
        for pareto in reversed(self.paretoLevels.values()):
            if pareto.ratioInLevel(ratio):
                level = pareto.level
        return level

    def getInfoHeaderColumn(self, col: str) -> str:
        href = f"https://wiki.bitplan.com/index.php/Truly_Tabular_RDF/Info#{col}"
        info = f"{col}<br><a href='{href}'style='color:white' target='_blank'>ⓘ</a>"
        return info

    def hasMinFrequency(self, record: dict) -> bool:
        """
        Check if the frequency of the given property record is greater than the minimal frequency

        Returns:
            True if property frequency is greater or equal than the minFrequency. Otherwise False
        """
        ok = float(record.get("%", 0)) >= self.minFrequency
        return ok

    def select(self) -> List[Tuple[str, dict]]:
        """
        select all properties that fulfill hasMinFrequency

        Returns:
            list of all selected properties as tuple list consisting of property id and record
        """
        selected = []
        for propertyId, propRecord in self.propertyMap.items():
            if self.hasMinFrequency(propRecord):
                selected.append((propertyId, propRecord))
        return selected

    def prepare(self):
        """
        prepare the propertyList

        Args:
            total(int): the total number of records
            paretoLevels(list): the pareto Levels to use
        """

        self.headerMap = {}
        cols = [
            "#",
            "%",
            "pareto",
            "property",
            "propertyId",
            "type",
            "1",
            "maxf",
            "nt",
            "nt%",
            "?f",
            "?ex",
            "✔",
        ]
        cols.extend(self.checkbox_cols)
        for col in cols:
            self.headerMap[col] = self.getInfoHeaderColumn(col)
        for i, prop in enumerate(self.propertyList):
            # add index as first column
            prop["#"] = i + 1
            prop.move_to_end("#", last=False)
            propLabel = prop.pop("propLabel")
            url = prop.pop("prop")
            itemId = url.replace("http://www.wikidata.org/entity/", "")
            prop["propertyId"] = itemId
            prop["property"] = Link.create(url, propLabel)
            prop["type"] = prop.pop("wbType").replace("http://wikiba.se/ontology#", "")
            prop["1"] = ""
            prop["maxf"] = ""
            prop["nt"] = ""
            prop["nt%"] = ""
            prop["?f"] = ""
            prop["?ex"] = ""
            prop["✔"] = ""
            # workaround count being first element
            prop["count"] = prop.pop("count")
            for col in self.checkbox_cols:
                prop[col] = False

            self.propertyMap[itemId] = prop

checkbox_cols: list property

get all my checkbox columns

__init__(inputList, total, paretoLevels, minFrequency)

Constructor

Parameters:

Name Type Description Default
propertyList(list)

the list of properties to show

 required
total(int)

total number of properties

 required
paretolLevels

a dict of paretoLevels with the key corresponding to the level

 required
minFrequency(float)

the minimum frequency of the properties to select in percent

 required
Source code in wd/truly_tabular_display.py 
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
def __init__(
    self,
    inputList,
    total: int,
    paretoLevels: Dict[int, Pareto],
    minFrequency: float,
):
    """
       Constructor

    Args:
        propertyList(list): the list of properties to show
        total(int): total number of properties
        paretolLevels: a dict of paretoLevels with the key corresponding to the level
        minFrequency(float): the minimum frequency of the properties to select in percent
    """
    self.propertyMap: Dict[str, dict] = dict()
    self.headerMap = {}
    self.propertyList = []
    self.total = total
    self.paretoLevels = paretoLevels
    self.minFrequency = minFrequency
    for record in inputList:
        ratio = int(record["count"]) / self.total
        level = self.getParetoLevel(ratio)
        record["%"] = f"{ratio*100:.1f}"
        record["pareto"] = level
        # if record["pareto"]<=paretoLimit:
        orecord = collections.OrderedDict(record.copy())
        self.propertyList.append(orecord)
    pass

hasMinFrequency(record)

Check if the frequency of the given property record is greater than the minimal frequency

Returns:

Type Description
bool

True if property frequency is greater or equal than the minFrequency. Otherwise False
Source code in wd/truly_tabular_display.py 
165
166
167
168
169
170
171
172
173
def hasMinFrequency(self, record: dict) -> bool:
    """
    Check if the frequency of the given property record is greater than the minimal frequency

    Returns:
        True if property frequency is greater or equal than the minFrequency. Otherwise False
    """
    ok = float(record.get("%", 0)) >= self.minFrequency
    return ok

prepare()

prepare the propertyList

Parameters:

Name Type Description Default
total(int)

the total number of records

 required
paretoLevels(list)

the pareto Levels to use

 required
Source code in wd/truly_tabular_display.py 
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
def prepare(self):
    """
    prepare the propertyList

    Args:
        total(int): the total number of records
        paretoLevels(list): the pareto Levels to use
    """

    self.headerMap = {}
    cols = [
        "#",
        "%",
        "pareto",
        "property",
        "propertyId",
        "type",
        "1",
        "maxf",
        "nt",
        "nt%",
        "?f",
        "?ex",
        "✔",
    ]
    cols.extend(self.checkbox_cols)
    for col in cols:
        self.headerMap[col] = self.getInfoHeaderColumn(col)
    for i, prop in enumerate(self.propertyList):
        # add index as first column
        prop["#"] = i + 1
        prop.move_to_end("#", last=False)
        propLabel = prop.pop("propLabel")
        url = prop.pop("prop")
        itemId = url.replace("http://www.wikidata.org/entity/", "")
        prop["propertyId"] = itemId
        prop["property"] = Link.create(url, propLabel)
        prop["type"] = prop.pop("wbType").replace("http://wikiba.se/ontology#", "")
        prop["1"] = ""
        prop["maxf"] = ""
        prop["nt"] = ""
        prop["nt%"] = ""
        prop["?f"] = ""
        prop["?ex"] = ""
        prop["✔"] = ""
        # workaround count being first element
        prop["count"] = prop.pop("count")
        for col in self.checkbox_cols:
            prop[col] = False

        self.propertyMap[itemId] = prop

select()

select all properties that fulfill hasMinFrequency

Returns:

Type Description
List[Tuple[str, dict]]

list of all selected properties as tuple list consisting of property id and record
Source code in wd/truly_tabular_display.py 
175
176
177
178
179
180
181
182
183
184
185
186
def select(self) -> List[Tuple[str, dict]]:
    """
    select all properties that fulfill hasMinFrequency

    Returns:
        list of all selected properties as tuple list consisting of property id and record
    """
    selected = []
    for propertyId, propRecord in self.propertyMap.items():
        if self.hasMinFrequency(propRecord):
            selected.append((propertyId, propRecord))
    return selected

TrulyTabularConfig dataclass

Configuration class for Truly Tabular operations.

Attributes:

Name Type Description
lang str

Language code (default is "en").
list_separator str

Character used to separate items in lists (default is "|").
endpoint_name str

Name of the endpoint to use (default is "wikidata").
Source code in wd/truly_tabular_display.py 
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
@dataclass
class TrulyTabularConfig:
    """
    Configuration class for Truly Tabular operations.

    Attributes:
        lang (str): Language code (default is "en").
        list_separator (str): Character used to separate items in lists (default is "|").
        endpoint_name (str): Name of the endpoint to use (default is "wikidata").
    """

    lang: str = "en"
    list_separator: str = "|"
    endpoint_name: str = "wikidata"
    pareto_level = 1
    # minimum percentual frequency of availability
    min_property_frequency = 20.0

    def __post_init__(self):
        """
        Post-initialization to setup additional attributes.
        """
        self.endpoints = EndpointManager.getEndpoints(lang="sparql")
        self.languages = Lang.get_language_dict()
        self.pareto_levels = {}
        self.pareto_select = {}
        for level in range(1, 10):
            pareto = Pareto(level)
            self.pareto_levels[level] = pareto
            self.pareto_select[level] = pareto.asText(long=True)
        pass

    @property
    def sparql_endpoint(self) -> Endpoint:
        endpoint = self.endpoints.get(self.endpoint_name, None)
        return endpoint

    @property
    def pareto(self) -> Pareto:
        pareto = self.pareto_levels[self.pareto_level]
        return pareto

    def setup_ui(self, webserver):
        """
        setup the user interface
        """
        with ui.grid(columns=2):
            webserver.add_select("lang", self.languages, with_input=True).bind_value(
                self, "lang"
            )
            list_separators = {
                "|": "|",
                ",": ",",
                ";": ";",
                ":": ":",
                "\x1c": "FS - ASCII(28)",
                "\x1d": "GS - ASCII(29)",
                "\x1e": "RS - ASCII(30)",
                "\x1f": "US - ASCII(31)",
            }
            webserver.add_select("List separator", list_separators).bind_value(
                self, "list_separator"
            )
            webserver.add_select("Endpoint", list(self.endpoints.keys())).bind_value(
                self, "endpoint_name"
            )
            webserver.add_select("Pareto level", self.pareto_select).bind_value(
                self, "pareto_level"
            )

__post_init__()

Post-initialization to setup additional attributes.

Source code in wd/truly_tabular_display.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
def __post_init__(self):
    """
    Post-initialization to setup additional attributes.
    """
    self.endpoints = EndpointManager.getEndpoints(lang="sparql")
    self.languages = Lang.get_language_dict()
    self.pareto_levels = {}
    self.pareto_select = {}
    for level in range(1, 10):
        pareto = Pareto(level)
        self.pareto_levels[level] = pareto
        self.pareto_select[level] = pareto.asText(long=True)
    pass

setup_ui(webserver)

setup the user interface

Source code in wd/truly_tabular_display.py 
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
def setup_ui(self, webserver):
    """
    setup the user interface
    """
    with ui.grid(columns=2):
        webserver.add_select("lang", self.languages, with_input=True).bind_value(
            self, "lang"
        )
        list_separators = {
            "|": "|",
            ",": ",",
            ";": ";",
            ":": ":",
            "\x1c": "FS - ASCII(28)",
            "\x1d": "GS - ASCII(29)",
            "\x1e": "RS - ASCII(30)",
            "\x1f": "US - ASCII(31)",
        }
        webserver.add_select("List separator", list_separators).bind_value(
            self, "list_separator"
        )
        webserver.add_select("Endpoint", list(self.endpoints.keys())).bind_value(
            self, "endpoint_name"
        )
        webserver.add_select("Pareto level", self.pareto_select).bind_value(
            self, "pareto_level"
        )

TrulyTabularDisplay

Displays a truly tabular analysis for a given Wikidata item

Source code in wd/truly_tabular_display.py 
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
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
class TrulyTabularDisplay:
    """
    Displays a truly tabular analysis for a given Wikidata
    item
    """

    def __init__(self, solution: "WdgridSolution", qid: str):
        """
        constructor
        """
        self.solution = solution
        self.config = solution.tt_config
        self.search_predicate = "wdt:P31"
        self.qid = qid
        self.tt = None
        self.naive_query_view = None
        self.aggregate_query_view = None
        self.setup()

    async def ui_yield(self):
        await asyncio.sleep(0)  # allow other tasks to run on the event loop

    @staticmethod
    def isTimeoutException(ex: EndPointInternalError):
        """
        Checks if the given exception is a query timeout exception

        Returns:
            True if the given exception is caused by a query timeout
        """
        check_for = "java.util.concurrent.TimeoutException"
        msg = ex.args[0]
        res = False
        if isinstance(msg, str):
            if check_for in msg:
                res = True
        return res

    def setup(self):
        """
        set up the user interface
        """
        with ui.element("div").classes("w-full") as self.main_container:
            with ui.splitter() as splitter:
                with splitter.before:
                    with ui.row() as self.sp_row:
                        self.item_input = ui.input(
                            "item", value=self.qid, on_change=self.update_display
                        ).bind_value(self, "qid")
                        predicates = {
                            "wdt:P31": "instance of",
                            "wdt:P31/wdt:P279*": "subclass of",
                            "wdt:P179": "part of the series",
                        }
                        self.solution.add_select(
                            "predicate",
                            predicates,
                            with_input=True,
                            value=self.search_predicate,
                            on_change=self.update_display,
                        ).bind_value(self, "search_predicate")

                    with ui.row() as self.item_row:
                        self.item_link_view = ui.html()
                        self.item_count_view = ui.html()
                    with ui.row():
                        self.solution.add_select(
                            "Pareto level",
                            self.config.pareto_select,
                            on_change=self.on_pareto_change,
                        ).bind_value(self.config, "pareto_level")
                        self.min_property_frequency_input = ui.input(
                            "min%",
                            value=str(self.config.min_property_frequency),
                        ).on("keydown.enter", self.on_min_property_frequency_change)
                with splitter.after as self.query_display_container:
                    self.count_query_view = QueryView(
                        self.solution,
                        name="count Query",
                        sparql_endpoint=self.config.sparql_endpoint,
                    )
                    self.property_query_view = QueryView(
                        self.solution,
                        name="property Query",
                        sparql_endpoint=self.config.sparql_endpoint,
                    )
            with ui.row() as self.generate_button_row:
                self.generate_button = ui.button(
                    "Generate SPARQL queries", on_click=self.on_generate_button_click
                )
                self.generate_button.disable()
            with ui.row() as self.progressbar_row:
                self.progress_bar = NiceguiProgressbar(
                    total=0, desc="Property statistics", unit="prop"
                )
            with ui.row() as self.property_grid_row:
                config = GridConfig(multiselect=True)
                self.property_grid = ListOfDictsGrid(config=config)
        # immediately do an async call of update view
        # ui.timer(0, self.update_display, once=True)

    def createTrulyTabular(self, itemQid: str, propertyIds=[]):
        """
        create a Truly Tabular configuration for my configure endpoint and the given itemQid and
        propertyIds

        Args:
            itemQid(str): e.g. Q5 human
            propertyIds(list): list of property Ids (if any) such as P17 country
        """
        tt = TrulyTabular(
            itemQid=itemQid,
            propertyIds=propertyIds,
            search_predicate=self.search_predicate,
            endpointConf=self.config.sparql_endpoint,
            debug=self.solution.debug,
        )
        return tt

    def wikiTrulyTabularPropertyStats(self, itemId: str, propertyId: str):
        """
        get the truly tabular property statistics

        Args:
            itemId(str): the Wikidata item identifier
            propertyId(str): the property id
        """
        try:
            tt = self.createTrulyTabular(itemId, propertyIds=[propertyId])
            statsRow = next(tt.genPropertyStatistics())
            for key in ["queryf", "queryex"]:
                queryText = statsRow[key]
                sparql = f"# This query was generated by Truly Tabular\n{queryText}"
                query = Query(name=key, query=sparql)
                tryItUrlEncoded = query.getTryItUrl(
                    baseurl=self.config.sparql_endpoint.website,
                    database=self.config.sparql_endpoint.database,
                )
                tryItLink = Link.create(
                    url=tryItUrlEncoded,
                    text="try it!",
                    tooltip=f"try out with {self.config.sparql_endpoint.name}",
                    target="_blank",
                )
                statsRow[f"{key}TryIt"] = tryItLink
            return statsRow
        except (BaseException, HTTPError) as ex:
            self.solution.handle_exception(ex)
            return None

    async def getPropertyIdMap(self) -> Dict:
        """
        get the map of selected property ids
        with generation specs

        Returns:
            dict: a dict of list
        """
        idMap = {}
        cols = self.property_selection.checkbox_cols
        selected_rows = await self.property_grid.get_selected_rows()
        for srow in selected_rows:
            propertyId = srow["propertyId"]
            key_value = srow["#"]
            genList = []
            for col_key in cols:
                checked = self.property_grid.get_cell_value(key_value, col_key)
                if checked:
                    genList.append(col_key)
            idMap[propertyId] = genList
        return idMap

    async def generateQueries(self):
        """
        generate and show the queries
        """
        try:
            propertyIdMap = await self.getPropertyIdMap()
            tt = self.createTrulyTabular(
                itemQid=self.qid, propertyIds=list(propertyIdMap.keys())
            )

            if self.naive_query_view is None:
                with self.query_display_container:
                    self.naive_query_view = QueryView(
                        self.solution,
                        name="naive Query",
                        sparql_endpoint=self.config.sparql_endpoint,
                    )
            if self.aggregate_query_view is None:
                with self.query_display_container:
                    self.aggregate_query_view = QueryView(
                        self.solution,
                        name="aggregate Query",
                        sparql_endpoint=self.config.sparql_endpoint,
                    )
            sparqlQuery = tt.generateSparqlQuery(
                genMap=propertyIdMap,
                naive=True,
                lang=self.config.lang,
                listSeparator=self.config.list_separator,
            )
            naiveSparqlQuery = Query(name="naive SPARQL Query", query=sparqlQuery)
            self.naive_query_view.show_query(naiveSparqlQuery.query)
            sparqlQuery = tt.generateSparqlQuery(
                genMap=propertyIdMap,
                naive=False,
                lang=self.config.lang,
                listSeparator=self.config.list_separator,
            )
            self.aggregateSparqlQuery = Query(
                name="aggregate SPARQL Query", query=sparqlQuery
            )
            self.aggregate_query_view.show_query(self.aggregateSparqlQuery.query)
            ui.notify("SPARQL queries generated")
        except Exception as ex:
            self.solution.handle_exception(ex)

    async def on_generate_button_click(self, _event):
        """
        handle the generate button click
        """
        try:
            ui.notify(f"generating SPARQL query for {str(self.tt)}")
            await self.generateQueries()
        except BaseException as ex:
            self.solution.handle_exception(ex)

    async def on_min_property_frequency_change(self, _event):
        """
        handle a change in the minimum property frequency input
        """
        value_str = self.min_property_frequency_input.value
        try:
            self.config.min_property_frequency = float(value_str)
            ui.notify(f"new freq: {self.config.min_property_frequency}")
            await self.update_display()
        except Exception as _ex:
            ui.notify(f"invalid frequency value {value_str}")
            pass

    async def on_pareto_change(self, _event):
        """
        handle changes in the pareto level
        """
        ui.notify(f"pareto level changed to {self.config.pareto_level} ")
        self.config.min_property_frequency = self.config.pareto.asPercent()
        self.min_property_frequency_input.value = str(
            self.config.min_property_frequency
        )

    def get_stats_rows(self, property_grid_rows: list):
        """
        get the statistic rows for the given property_grid_rows
        """
        for row in property_grid_rows:
            property_id = row["propertyId"]
            row_key = row["#"]
            stats_row = self.wikiTrulyTabularPropertyStats(self.tt.itemQid, property_id)
            if stats_row:
                stats_row["✔"] = "✔"
            else:
                stats_row = {"✔": "❌"}
            for col_key, statsColumn in [
                ("1", "1"),
                ("maxf", "maxf"),
                ("nt", "non tabular"),
                ("nt%", "non tabular%"),
                ("?f", "queryfTryIt"),
                ("?ex", "queryexTryIt"),
                ("✔", "✔"),
            ]:
                if statsColumn in stats_row:
                    value = stats_row[statsColumn]
                    self.property_grid.update_cell(row_key, col_key, value)
            self.property_grid.update()
            pass

    def update_item_count_view(self):
        """
        update the item count
        """
        try:
            self.ttcount, countQuery = self.tt.count()
            self.count_query_view.show_query(countQuery)
            content = "❓" if self.tt.error else f"{self.ttcount} instances found"
            with self.item_row:
                self.item_count_view.content = content
            if not self.tt.error:
                self.update_property_query_view(total=self.ttcount)

        except Exception as ex:
            self.solution.handle_exception(ex)

    def update_property_query_view(self, total: int):
        """
        update the property query view
        """
        try:
            pareto = self.config.pareto
            if total is not None:
                min_count = round(total * self.config.min_property_frequency / 100.0)
            else:
                min_count = 0
            msg = f"searching properties with at least {min_count} usages"
            with self.main_container:
                ui.notify(msg)
            mfp_query = self.tt.mostFrequentPropertiesQuery(minCount=min_count)
            self.property_query_view.show_query(mfp_query.query)
            self.update_properties_table(mfp_query)
        except Exception as ex:
            self.solution.handle_exception(ex)

    def prepare_generation_specs(self):
        """
        prepare the interactive generation specification
        """
        # render generation spec columns as checkboxes
        for col in self.property_selection.checkbox_cols:
            self.property_grid.set_checkbox_renderer(col)
            pass
        for row in self.property_selection.propertyList:
            has_min_frequency = self.property_selection.hasMinFrequency(row)
            row["count"] = True
            if has_min_frequency:
                if row["type"] == "WikibaseItem":
                    row["label"] = True
            else:
                row["ignore"] = True
            pass
        col_def = self.property_grid.get_column_def("#")
        col_def["headerCheckboxSelection"] = True
        self.property_grid.update()
        self.property_grid.select_all_rows()
        self.generate_button.enable()

    def update_properties_table(self, mfp_query):
        """
        update my properties table

        Args:
            mfp_query(Query): the query for the most frequently used properties
        """
        try:
            with self.query_display_container:
                msg = f"running query for most frequently used properties of {str(self.tt)} ..."
                ui.notify(msg)
            try:
                property_lod = self.tt.sparql.queryAsListOfDicts(mfp_query.query)
            except EndPointInternalError as ex:
                if self.isTimeoutException(ex):
                    raise Exception("Query timeout of the property table query")
            self.property_selection = PropertySelection(
                property_lod,
                total=self.ttcount,
                paretoLevels=self.config.pareto_levels,
                minFrequency=self.config.min_property_frequency,
            )
            self.property_selection.prepare()
            with self.property_grid_row:
                self.view_lod = self.property_selection.propertyList
                self.property_grid.load_lod(self.view_lod)
                self.property_grid.set_checkbox_selection("#")
                self.property_grid.update()
            self.update_property_stats()
            self.prepare_generation_specs()
        except Exception as ex:
            self.solution.handle_exception(ex)

    def update_property_stats(self):
        """
        update the property statistics
        """
        try:
            count = len(self.property_selection.propertyList)
            with self.main_container:
                ui.notify(f"Getting property statistics for {count} properties")
                self.progress_bar.total = count
                self.progress_bar.reset()
            for row in self.property_selection.propertyList:
                # run in background
                asyncio.run(run.io_bound(self.get_stats_rows, [row]))
                with self.main_container:
                    self.progress_bar.update(1)
            pass
            with self.main_container:
                self.progress_bar.reset()
                ui.notify(f"Done getting statistics for {count} properties")
        except Exception as ex:
            self.solution.handle_exception(ex)

    async def on_property_grid_selection_change(self, event):
        """
        the property grid selection has changed
        """
        source = event.args.get("source", None)
        if source == "checkboxSelected":
            selected_rows = await self.property_grid.get_selected_rows()
            ui.notify(f"Selection changed: {selected_rows}")

    async def update_item_link_view(self):
        with self.item_row:
            item_text = self.tt.item.asText(long=True)
            item_url = self.tt.item.url
            item_link = Link.create(item_url, item_text)
            self.item_link_view.content = item_link

    async def update_display(self):
        """
        update the display
        """
        try:
            if self.solution.log_view:
                self.solution.log_view.clear()
            self.tt = self.createTrulyTabular(self.qid)
            for query_view in self.count_query_view, self.property_query_view:
                query_view.sparql_endpoint = self.config.sparql_endpoint
            # Initialize TrulyTabular with the qid
            await self.update_item_link_view()
            await run.io_bound(self.update_item_count_view)
        except Exception as ex:
            self.solution.handle_exception(ex)

__init__(solution, qid)

constructor

Source code in wd/truly_tabular_display.py 
247
248
249
250
251
252
253
254
255
256
257
258
def __init__(self, solution: "WdgridSolution", qid: str):
    """
    constructor
    """
    self.solution = solution
    self.config = solution.tt_config
    self.search_predicate = "wdt:P31"
    self.qid = qid
    self.tt = None
    self.naive_query_view = None
    self.aggregate_query_view = None
    self.setup()

createTrulyTabular(itemQid, propertyIds=[])

create a Truly Tabular configuration for my configure endpoint and the given itemQid and propertyIds

Parameters:

Name Type Description Default
itemQid(str)

e.g. Q5 human

 required
propertyIds(list)

list of property Ids (if any) such as P17 country

 required
Source code in wd/truly_tabular_display.py 
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
def createTrulyTabular(self, itemQid: str, propertyIds=[]):
    """
    create a Truly Tabular configuration for my configure endpoint and the given itemQid and
    propertyIds

    Args:
        itemQid(str): e.g. Q5 human
        propertyIds(list): list of property Ids (if any) such as P17 country
    """
    tt = TrulyTabular(
        itemQid=itemQid,
        propertyIds=propertyIds,
        search_predicate=self.search_predicate,
        endpointConf=self.config.sparql_endpoint,
        debug=self.solution.debug,
    )
    return tt

generateQueries() async

generate and show the queries

Source code in wd/truly_tabular_display.py 
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
async def generateQueries(self):
    """
    generate and show the queries
    """
    try:
        propertyIdMap = await self.getPropertyIdMap()
        tt = self.createTrulyTabular(
            itemQid=self.qid, propertyIds=list(propertyIdMap.keys())
        )

        if self.naive_query_view is None:
            with self.query_display_container:
                self.naive_query_view = QueryView(
                    self.solution,
                    name="naive Query",
                    sparql_endpoint=self.config.sparql_endpoint,
                )
        if self.aggregate_query_view is None:
            with self.query_display_container:
                self.aggregate_query_view = QueryView(
                    self.solution,
                    name="aggregate Query",
                    sparql_endpoint=self.config.sparql_endpoint,
                )
        sparqlQuery = tt.generateSparqlQuery(
            genMap=propertyIdMap,
            naive=True,
            lang=self.config.lang,
            listSeparator=self.config.list_separator,
        )
        naiveSparqlQuery = Query(name="naive SPARQL Query", query=sparqlQuery)
        self.naive_query_view.show_query(naiveSparqlQuery.query)
        sparqlQuery = tt.generateSparqlQuery(
            genMap=propertyIdMap,
            naive=False,
            lang=self.config.lang,
            listSeparator=self.config.list_separator,
        )
        self.aggregateSparqlQuery = Query(
            name="aggregate SPARQL Query", query=sparqlQuery
        )
        self.aggregate_query_view.show_query(self.aggregateSparqlQuery.query)
        ui.notify("SPARQL queries generated")
    except Exception as ex:
        self.solution.handle_exception(ex)

getPropertyIdMap() async

get the map of selected property ids with generation specs

Returns:

Name Type Description
dict Dict

a dict of list
Source code in wd/truly_tabular_display.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
async def getPropertyIdMap(self) -> Dict:
    """
    get the map of selected property ids
    with generation specs

    Returns:
        dict: a dict of list
    """
    idMap = {}
    cols = self.property_selection.checkbox_cols
    selected_rows = await self.property_grid.get_selected_rows()
    for srow in selected_rows:
        propertyId = srow["propertyId"]
        key_value = srow["#"]
        genList = []
        for col_key in cols:
            checked = self.property_grid.get_cell_value(key_value, col_key)
            if checked:
                genList.append(col_key)
        idMap[propertyId] = genList
    return idMap

get_stats_rows(property_grid_rows)

get the statistic rows for the given property_grid_rows

Source code in wd/truly_tabular_display.py 
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
def get_stats_rows(self, property_grid_rows: list):
    """
    get the statistic rows for the given property_grid_rows
    """
    for row in property_grid_rows:
        property_id = row["propertyId"]
        row_key = row["#"]
        stats_row = self.wikiTrulyTabularPropertyStats(self.tt.itemQid, property_id)
        if stats_row:
            stats_row["✔"] = "✔"
        else:
            stats_row = {"✔": "❌"}
        for col_key, statsColumn in [
            ("1", "1"),
            ("maxf", "maxf"),
            ("nt", "non tabular"),
            ("nt%", "non tabular%"),
            ("?f", "queryfTryIt"),
            ("?ex", "queryexTryIt"),
            ("✔", "✔"),
        ]:
            if statsColumn in stats_row:
                value = stats_row[statsColumn]
                self.property_grid.update_cell(row_key, col_key, value)
        self.property_grid.update()
        pass

isTimeoutException(ex) staticmethod

Checks if the given exception is a query timeout exception

Returns:

Type Description

True if the given exception is caused by a query timeout
Source code in wd/truly_tabular_display.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
@staticmethod
def isTimeoutException(ex: EndPointInternalError):
    """
    Checks if the given exception is a query timeout exception

    Returns:
        True if the given exception is caused by a query timeout
    """
    check_for = "java.util.concurrent.TimeoutException"
    msg = ex.args[0]
    res = False
    if isinstance(msg, str):
        if check_for in msg:
            res = True
    return res

on_generate_button_click(_event) async

handle the generate button click

Source code in wd/truly_tabular_display.py 
459
460
461
462
463
464
465
466
467
async def on_generate_button_click(self, _event):
    """
    handle the generate button click
    """
    try:
        ui.notify(f"generating SPARQL query for {str(self.tt)}")
        await self.generateQueries()
    except BaseException as ex:
        self.solution.handle_exception(ex)

on_min_property_frequency_change(_event) async

handle a change in the minimum property frequency input

Source code in wd/truly_tabular_display.py 
469
470
471
472
473
474
475
476
477
478
479
480
async def on_min_property_frequency_change(self, _event):
    """
    handle a change in the minimum property frequency input
    """
    value_str = self.min_property_frequency_input.value
    try:
        self.config.min_property_frequency = float(value_str)
        ui.notify(f"new freq: {self.config.min_property_frequency}")
        await self.update_display()
    except Exception as _ex:
        ui.notify(f"invalid frequency value {value_str}")
        pass

on_pareto_change(_event) async

handle changes in the pareto level

Source code in wd/truly_tabular_display.py 
482
483
484
485
486
487
488
489
490
async def on_pareto_change(self, _event):
    """
    handle changes in the pareto level
    """
    ui.notify(f"pareto level changed to {self.config.pareto_level} ")
    self.config.min_property_frequency = self.config.pareto.asPercent()
    self.min_property_frequency_input.value = str(
        self.config.min_property_frequency
    )

on_property_grid_selection_change(event) async

the property grid selection has changed

Source code in wd/truly_tabular_display.py 
632
633
634
635
636
637
638
639
async def on_property_grid_selection_change(self, event):
    """
    the property grid selection has changed
    """
    source = event.args.get("source", None)
    if source == "checkboxSelected":
        selected_rows = await self.property_grid.get_selected_rows()
        ui.notify(f"Selection changed: {selected_rows}")

prepare_generation_specs()

prepare the interactive generation specification

Source code in wd/truly_tabular_display.py 
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
def prepare_generation_specs(self):
    """
    prepare the interactive generation specification
    """
    # render generation spec columns as checkboxes
    for col in self.property_selection.checkbox_cols:
        self.property_grid.set_checkbox_renderer(col)
        pass
    for row in self.property_selection.propertyList:
        has_min_frequency = self.property_selection.hasMinFrequency(row)
        row["count"] = True
        if has_min_frequency:
            if row["type"] == "WikibaseItem":
                row["label"] = True
        else:
            row["ignore"] = True
        pass
    col_def = self.property_grid.get_column_def("#")
    col_def["headerCheckboxSelection"] = True
    self.property_grid.update()
    self.property_grid.select_all_rows()
    self.generate_button.enable()

setup()

set up the user interface

Source code in wd/truly_tabular_display.py 
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
def setup(self):
    """
    set up the user interface
    """
    with ui.element("div").classes("w-full") as self.main_container:
        with ui.splitter() as splitter:
            with splitter.before:
                with ui.row() as self.sp_row:
                    self.item_input = ui.input(
                        "item", value=self.qid, on_change=self.update_display
                    ).bind_value(self, "qid")
                    predicates = {
                        "wdt:P31": "instance of",
                        "wdt:P31/wdt:P279*": "subclass of",
                        "wdt:P179": "part of the series",
                    }
                    self.solution.add_select(
                        "predicate",
                        predicates,
                        with_input=True,
                        value=self.search_predicate,
                        on_change=self.update_display,
                    ).bind_value(self, "search_predicate")

                with ui.row() as self.item_row:
                    self.item_link_view = ui.html()
                    self.item_count_view = ui.html()
                with ui.row():
                    self.solution.add_select(
                        "Pareto level",
                        self.config.pareto_select,
                        on_change=self.on_pareto_change,
                    ).bind_value(self.config, "pareto_level")
                    self.min_property_frequency_input = ui.input(
                        "min%",
                        value=str(self.config.min_property_frequency),
                    ).on("keydown.enter", self.on_min_property_frequency_change)
            with splitter.after as self.query_display_container:
                self.count_query_view = QueryView(
                    self.solution,
                    name="count Query",
                    sparql_endpoint=self.config.sparql_endpoint,
                )
                self.property_query_view = QueryView(
                    self.solution,
                    name="property Query",
                    sparql_endpoint=self.config.sparql_endpoint,
                )
        with ui.row() as self.generate_button_row:
            self.generate_button = ui.button(
                "Generate SPARQL queries", on_click=self.on_generate_button_click
            )
            self.generate_button.disable()
        with ui.row() as self.progressbar_row:
            self.progress_bar = NiceguiProgressbar(
                total=0, desc="Property statistics", unit="prop"
            )
        with ui.row() as self.property_grid_row:
            config = GridConfig(multiselect=True)
            self.property_grid = ListOfDictsGrid(config=config)

update_display() async

update the display

Source code in wd/truly_tabular_display.py 
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
async def update_display(self):
    """
    update the display
    """
    try:
        if self.solution.log_view:
            self.solution.log_view.clear()
        self.tt = self.createTrulyTabular(self.qid)
        for query_view in self.count_query_view, self.property_query_view:
            query_view.sparql_endpoint = self.config.sparql_endpoint
        # Initialize TrulyTabular with the qid
        await self.update_item_link_view()
        await run.io_bound(self.update_item_count_view)
    except Exception as ex:
        self.solution.handle_exception(ex)

update_item_count_view()

update the item count

Source code in wd/truly_tabular_display.py 
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
def update_item_count_view(self):
    """
    update the item count
    """
    try:
        self.ttcount, countQuery = self.tt.count()
        self.count_query_view.show_query(countQuery)
        content = "❓" if self.tt.error else f"{self.ttcount} instances found"
        with self.item_row:
            self.item_count_view.content = content
        if not self.tt.error:
            self.update_property_query_view(total=self.ttcount)

    except Exception as ex:
        self.solution.handle_exception(ex)

update_properties_table(mfp_query)

update my properties table

Parameters:

Name Type Description Default
mfp_query(Query)

the query for the most frequently used properties

 required
Source code in wd/truly_tabular_display.py 
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
def update_properties_table(self, mfp_query):
    """
    update my properties table

    Args:
        mfp_query(Query): the query for the most frequently used properties
    """
    try:
        with self.query_display_container:
            msg = f"running query for most frequently used properties of {str(self.tt)} ..."
            ui.notify(msg)
        try:
            property_lod = self.tt.sparql.queryAsListOfDicts(mfp_query.query)
        except EndPointInternalError as ex:
            if self.isTimeoutException(ex):
                raise Exception("Query timeout of the property table query")
        self.property_selection = PropertySelection(
            property_lod,
            total=self.ttcount,
            paretoLevels=self.config.pareto_levels,
            minFrequency=self.config.min_property_frequency,
        )
        self.property_selection.prepare()
        with self.property_grid_row:
            self.view_lod = self.property_selection.propertyList
            self.property_grid.load_lod(self.view_lod)
            self.property_grid.set_checkbox_selection("#")
            self.property_grid.update()
        self.update_property_stats()
        self.prepare_generation_specs()
    except Exception as ex:
        self.solution.handle_exception(ex)

update_property_query_view(total)

update the property query view

Source code in wd/truly_tabular_display.py 
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
def update_property_query_view(self, total: int):
    """
    update the property query view
    """
    try:
        pareto = self.config.pareto
        if total is not None:
            min_count = round(total * self.config.min_property_frequency / 100.0)
        else:
            min_count = 0
        msg = f"searching properties with at least {min_count} usages"
        with self.main_container:
            ui.notify(msg)
        mfp_query = self.tt.mostFrequentPropertiesQuery(minCount=min_count)
        self.property_query_view.show_query(mfp_query.query)
        self.update_properties_table(mfp_query)
    except Exception as ex:
        self.solution.handle_exception(ex)

update_property_stats()

update the property statistics

Source code in wd/truly_tabular_display.py 
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
def update_property_stats(self):
    """
    update the property statistics
    """
    try:
        count = len(self.property_selection.propertyList)
        with self.main_container:
            ui.notify(f"Getting property statistics for {count} properties")
            self.progress_bar.total = count
            self.progress_bar.reset()
        for row in self.property_selection.propertyList:
            # run in background
            asyncio.run(run.io_bound(self.get_stats_rows, [row]))
            with self.main_container:
                self.progress_bar.update(1)
        pass
        with self.main_container:
            self.progress_bar.reset()
            ui.notify(f"Done getting statistics for {count} properties")
    except Exception as ex:
        self.solution.handle_exception(ex)

wikiTrulyTabularPropertyStats(itemId, propertyId)

get the truly tabular property statistics

Parameters:

Name Type Description Default
itemId(str)

the Wikidata item identifier

 required
propertyId(str)

the property id

 required
Source code in wd/truly_tabular_display.py 
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
def wikiTrulyTabularPropertyStats(self, itemId: str, propertyId: str):
    """
    get the truly tabular property statistics

    Args:
        itemId(str): the Wikidata item identifier
        propertyId(str): the property id
    """
    try:
        tt = self.createTrulyTabular(itemId, propertyIds=[propertyId])
        statsRow = next(tt.genPropertyStatistics())
        for key in ["queryf", "queryex"]:
            queryText = statsRow[key]
            sparql = f"# This query was generated by Truly Tabular\n{queryText}"
            query = Query(name=key, query=sparql)
            tryItUrlEncoded = query.getTryItUrl(
                baseurl=self.config.sparql_endpoint.website,
                database=self.config.sparql_endpoint.database,
            )
            tryItLink = Link.create(
                url=tryItUrlEncoded,
                text="try it!",
                tooltip=f"try out with {self.config.sparql_endpoint.name}",
                target="_blank",
            )
            statsRow[f"{key}TryIt"] = tryItLink
        return statsRow
    except (BaseException, HTTPError) as ex:
        self.solution.handle_exception(ex)
        return None

version

Created on 2024-01-02

@author: wf

Version dataclass

Bases: object

Version handling for wdgrid

Source code in wd/version.py 
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
@dataclass
class Version(object):
    """
    Version handling for wdgrid
    """

    name = "wdgrid"
    version = wd.__version__
    description = "wikdata grid and sync"
    date = "2021-12-12"
    updated = "2024-03-07"

    authors = "Wolfgang Fahl"

    doc_url = "https://wiki.bitplan.com/index.php/Wdgrid"
    chat_url = "https://github.com/WolfgangFahl/wdgrid/discussions"
    cm_url = "https://github.com/WolfgangFahl/wdgrid"

    license = f"""Copyright 2023 contributors. All rights reserved.

  Licensed under the Apache License 2.0
  http://www.apache.org/licenses/LICENSE-2.0

  Distributed on an "AS IS" basis without warranties
  or conditions of any kind, either express or implied."""

    longDescription = f"""{name} version {version}
{description}

  Created by {authors} on {date} last updated {updated}"""

wdgrid_cmd

Created on 2024-01-03

@author: wf

WdgridCmd

Bases: WebserverCmd

Command line for wiki data grid web server

Source code in wd/wdgrid_cmd.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class WdgridCmd(WebserverCmd):
    """
    Command line for wiki data grid web server
    """

    def getArgParser(self, description: str, version_msg) -> ArgumentParser:
        """
        override the default argparser call
        """
        parser = super().getArgParser(description, version_msg)
        parser.add_argument(
            "-en",
            "--endpointName",
            default="wikidata",
            help=f"Name of the endpoint to use for queries. Available by default: {EndpointManager.getEndpointNames(lang='sparql')}",
        )
        return parser

getArgParser(description, version_msg)

override the default argparser call

Source code in wd/wdgrid_cmd.py 
21
22
23
24
25
26
27
28
29
30
31
32
def getArgParser(self, description: str, version_msg) -> ArgumentParser:
    """
    override the default argparser call
    """
    parser = super().getArgParser(description, version_msg)
    parser.add_argument(
        "-en",
        "--endpointName",
        default="wikidata",
        help=f"Name of the endpoint to use for queries. Available by default: {EndpointManager.getEndpointNames(lang='sparql')}",
    )
    return parser

main(argv=None)

main call

Source code in wd/wdgrid_cmd.py 
35
36
37
38
39
40
41
def main(argv: list = None):
    """
    main call
    """
    cmd = WdgridCmd(config=WdgridWebServer.get_config(), webserver_cls=WdgridWebServer)
    exit_code = cmd.cmd_main(argv)
    return exit_code

Created on 2024-01-03

@author: wf

WikidataItemSearch

wikidata item search

Source code in wd/wditem_search.py 
 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
class WikidataItemSearch:
    """
    wikidata item search
    """

    def __init__(self, webserver: NiceGuiWebserver, record_filter: Callable = None):
        """
        Initialize the WikidataItemSearch with the given webserver.

        Args:
            webserver (NiceGuiWebserver): The webserver to attach the search UI.
            record_filter(Callable): callback for displayed found records
        """
        self.webserver = webserver
        self.record_filter = record_filter
        self.limit = 9
        self.wd_search = WikidataSearch(self.webserver.tt_config.lang)
        self.search_debounce_task = None
        self.keyStrokeTime = 0.65  # minimum time in seconds to wait between keystrokes before starting searching
        self.search_result_row = None
        self.setup()

    def setup(self):
        """
        setup the user interface
        """
        with ui.card().style("width: 25%"):
            with ui.grid(rows=1, columns=4):
                ui.label("limit:")
                self.limit_slider = (
                    ui.slider(min=2, max=50, value=self.limit)
                    .props("label-always")
                    .bind_value(self, "limit")
                )
            with ui.row():
                self.search_input = ui.input(
                    label="search", on_change=self.on_search_change
                ).props("size=80")
        with ui.row() as self.search_result_row:
            self.search_result_grid = ListOfDictsGrid()

    async def on_search_change(self, _args):
        """
        react on changes in the search input
        """
        # Cancel the existing search task if it's still waiting
        if self.search_debounce_task:
            self.search_debounce_task.cancel()

        # Create a new task for the new search
        self.search_debounce_task = asyncio.create_task(self.debounced_search())

    async def debounced_search(self):
        """
        Waits for a period of inactivity and then performs the search.
        """
        try:
            # Wait for the debounce period (keyStrokeTime)
            await asyncio.sleep(self.keyStrokeTime)
            search_for = self.search_input.value
            if self.search_result_row:
                with self.search_result_row:
                    lang = self.webserver.tt_config.lang
                    ui.notify(f"searching wikidata for {search_for} ({lang})...")
                    self.wd_search.language = lang
                    wd_search_result = self.wd_search.searchOptions(
                        search_for, limit=self.limit
                    )
                    view_lod = self.get_selection_view_lod(wd_search_result)
                    self.search_result_grid.load_lod(view_lod)
                    # self.search_result_grid.set_checkbox_selection("#")
                    self.search_result_grid.update()
        except asyncio.CancelledError:
            # The search was cancelled because of new input, so just quietly exit
            pass
        except BaseException as ex:
            self.webserver.handle_exception(ex, self.webserver)

    def get_selection_view_lod(self, wd_search_result: list) -> dict:
        """
        Convert the Wikidata search result list of dict to a selection.

        Args:
            wd_search_result (List[Dict[str, Any]]): The search results from Wikidata.

        Returns:
            List[Dict[str, Any]]: The list of dictionaries formatted for view.
        """
        view_lod = []
        for qid, itemLabel, desc in wd_search_result:
            url = f"https://www.wikidata.org/wiki/{qid}"
            link = Link.create(url, qid)
            row = {
                "#": len(view_lod) + 1,
                "qid": link,
                "label": itemLabel,
                "desc": desc,
            }
            if self.record_filter:
                self.record_filter(qid, row)
            view_lod.append(row)
        return view_lod

__init__(webserver, record_filter=None)

Initialize the WikidataItemSearch with the given webserver.

Parameters:

Name Type Description Default
webserver NiceGuiWebserver

The webserver to attach the search UI.

 required
record_filter(Callable)

callback for displayed found records

 required
Source code in wd/wditem_search.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(self, webserver: NiceGuiWebserver, record_filter: Callable = None):
    """
    Initialize the WikidataItemSearch with the given webserver.

    Args:
        webserver (NiceGuiWebserver): The webserver to attach the search UI.
        record_filter(Callable): callback for displayed found records
    """
    self.webserver = webserver
    self.record_filter = record_filter
    self.limit = 9
    self.wd_search = WikidataSearch(self.webserver.tt_config.lang)
    self.search_debounce_task = None
    self.keyStrokeTime = 0.65  # minimum time in seconds to wait between keystrokes before starting searching
    self.search_result_row = None
    self.setup()

Waits for a period of inactivity and then performs the search.

Source code in wd/wditem_search.py 
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
async def debounced_search(self):
    """
    Waits for a period of inactivity and then performs the search.
    """
    try:
        # Wait for the debounce period (keyStrokeTime)
        await asyncio.sleep(self.keyStrokeTime)
        search_for = self.search_input.value
        if self.search_result_row:
            with self.search_result_row:
                lang = self.webserver.tt_config.lang
                ui.notify(f"searching wikidata for {search_for} ({lang})...")
                self.wd_search.language = lang
                wd_search_result = self.wd_search.searchOptions(
                    search_for, limit=self.limit
                )
                view_lod = self.get_selection_view_lod(wd_search_result)
                self.search_result_grid.load_lod(view_lod)
                # self.search_result_grid.set_checkbox_selection("#")
                self.search_result_grid.update()
    except asyncio.CancelledError:
        # The search was cancelled because of new input, so just quietly exit
        pass
    except BaseException as ex:
        self.webserver.handle_exception(ex, self.webserver)

get_selection_view_lod(wd_search_result)

Convert the Wikidata search result list of dict to a selection.

Parameters:

Name Type Description Default
wd_search_result List[Dict[str, Any]]

The search results from Wikidata.

 required

Returns:

Type Description
dict

List[Dict[str, Any]]: The list of dictionaries formatted for view.
Source code in wd/wditem_search.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def get_selection_view_lod(self, wd_search_result: list) -> dict:
    """
    Convert the Wikidata search result list of dict to a selection.

    Args:
        wd_search_result (List[Dict[str, Any]]): The search results from Wikidata.

    Returns:
        List[Dict[str, Any]]: The list of dictionaries formatted for view.
    """
    view_lod = []
    for qid, itemLabel, desc in wd_search_result:
        url = f"https://www.wikidata.org/wiki/{qid}"
        link = Link.create(url, qid)
        row = {
            "#": len(view_lod) + 1,
            "qid": link,
            "label": itemLabel,
            "desc": desc,
        }
        if self.record_filter:
            self.record_filter(qid, row)
        view_lod.append(row)
    return view_lod

on_search_change(_args) async

react on changes in the search input

Source code in wd/wditem_search.py 
58
59
60
61
62
63
64
65
66
67
async def on_search_change(self, _args):
    """
    react on changes in the search input
    """
    # Cancel the existing search task if it's still waiting
    if self.search_debounce_task:
        self.search_debounce_task.cancel()

    # Create a new task for the new search
    self.search_debounce_task = asyncio.create_task(self.debounced_search())

setup()

setup the user interface

Source code in wd/wditem_search.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def setup(self):
    """
    setup the user interface
    """
    with ui.card().style("width: 25%"):
        with ui.grid(rows=1, columns=4):
            ui.label("limit:")
            self.limit_slider = (
                ui.slider(min=2, max=50, value=self.limit)
                .props("label-always")
                .bind_value(self, "limit")
            )
        with ui.row():
            self.search_input = ui.input(
                label="search", on_change=self.on_search_change
            ).props("size=80")
    with ui.row() as self.search_result_row:
        self.search_result_grid = ListOfDictsGrid()

webserver

Created on 2024-01-03

@author: wf

WdgridSolution

Bases: InputWebSolution

Wikidata Grid client specific UI

Source code in wd/webserver.py 
 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
class WdgridSolution(InputWebSolution):
    """
    Wikidata Grid client specific UI
    """

    def __init__(self, webserver: WdgridWebServer, client: Client):
        super().__init__(webserver, client)  # Call to the superclass constructor
        self.debug = webserver.debug
        self.tt_config = TrulyTabularConfig()

    async def truly_tabular(self, qid: str):
        """
        show a truly tabular analysis of the given Wikidata id

        Args:
            qid(str): the Wikidata id of the item to analyze
        """

        def show():
            self.ttd = TrulyTabularDisplay(self, qid)

        await self.setup_content_div(show)

    def configure_settings(self):
        """
        extra settings
        """
        self.tt_config.setup_ui(self)

    def prepare_ui(self):
        """
        overrideable configuration
        """
        self.tt_config.endpoint_name = self.args.endpointName

    async def home(self):
        """
        provide the main content page
        """

        def record_filter(qid: str, record: dict):
            if "label" and "desc" in record:
                text = f"""{record["label"]}({qid})☞{record["desc"]}"""
                tt_link = Link.create(f"/tt/{qid}", text)
                # getting the link to be at second position
                # is a bit tricky
                temp_items = list(record.items())
                # Add the new item in the second position
                temp_items.insert(1, ("truly tabular", tt_link))

                # Clear the original dictionary and update it with the new order of items
                record.clear()
                record.update(temp_items)

        def show():
            self.wd_item_search = WikidataItemSearch(self, record_filter=record_filter)

        await self.setup_content_div(show)

configure_settings()

extra settings

Source code in wd/webserver.py 
73
74
75
76
77
def configure_settings(self):
    """
    extra settings
    """
    self.tt_config.setup_ui(self)

home() async

provide the main content page

Source code in wd/webserver.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
async def home(self):
    """
    provide the main content page
    """

    def record_filter(qid: str, record: dict):
        if "label" and "desc" in record:
            text = f"""{record["label"]}({qid})☞{record["desc"]}"""
            tt_link = Link.create(f"/tt/{qid}", text)
            # getting the link to be at second position
            # is a bit tricky
            temp_items = list(record.items())
            # Add the new item in the second position
            temp_items.insert(1, ("truly tabular", tt_link))

            # Clear the original dictionary and update it with the new order of items
            record.clear()
            record.update(temp_items)

    def show():
        self.wd_item_search = WikidataItemSearch(self, record_filter=record_filter)

    await self.setup_content_div(show)

prepare_ui()

overrideable configuration

Source code in wd/webserver.py 
79
80
81
82
83
def prepare_ui(self):
    """
    overrideable configuration
    """
    self.tt_config.endpoint_name = self.args.endpointName

truly_tabular(qid) async

show a truly tabular analysis of the given Wikidata id

Parameters:

Name Type Description Default
qid(str)

the Wikidata id of the item to analyze

 required
Source code in wd/webserver.py 
60
61
62
63
64
65
66
67
68
69
70
71
async def truly_tabular(self, qid: str):
    """
    show a truly tabular analysis of the given Wikidata id

    Args:
        qid(str): the Wikidata id of the item to analyze
    """

    def show():
        self.ttd = TrulyTabularDisplay(self, qid)

    await self.setup_content_div(show)

WdgridWebServer

Bases: InputWebserver

Server for Wikidata Grid

Source code in wd/webserver.py 
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
class WdgridWebServer(InputWebserver):
    """
    Server for Wikidata Grid
    """

    @classmethod
    def get_config(cls) -> WebserverConfig:
        """
        get the configuration for this Webserver
        """
        copy_right = "(c)2022-2024 Wolfgang Fahl"
        config = WebserverConfig(
            short_name="wdgrid",
            copy_right=copy_right,
            version=Version(),
            default_port=9997,
        )
        server_config = WebserverConfig.get(config)
        server_config.solution_class = WdgridSolution
        return server_config

    def __init__(self):
        """Constructs all the necessary attributes for the WebServer object."""
        InputWebserver.__init__(self, config=WdgridWebServer.get_config())

        @ui.page("/tt/{qid}")
        async def truly_tabular(client: Client, qid: str):
            """
            initiate the truly tabular analysis for the given Wikidata QIDs
            """
            await self.page(client, WdgridSolution.truly_tabular, qid)

__init__()

Constructs all the necessary attributes for the WebServer object.

Source code in wd/webserver.py 
38
39
40
41
42
43
44
45
46
47
def __init__(self):
    """Constructs all the necessary attributes for the WebServer object."""
    InputWebserver.__init__(self, config=WdgridWebServer.get_config())

    @ui.page("/tt/{qid}")
    async def truly_tabular(client: Client, qid: str):
        """
        initiate the truly tabular analysis for the given Wikidata QIDs
        """
        await self.page(client, WdgridSolution.truly_tabular, qid)

get_config() classmethod

get the configuration for this Webserver

Source code in wd/webserver.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@classmethod
def get_config(cls) -> WebserverConfig:
    """
    get the configuration for this Webserver
    """
    copy_right = "(c)2022-2024 Wolfgang Fahl"
    config = WebserverConfig(
        short_name="wdgrid",
        copy_right=copy_right,
        version=Version(),
        default_port=9997,
    )
    server_config = WebserverConfig.get(config)
    server_config.solution_class = WdgridSolution
    return server_config