Skip to content

Utilities

This module provides the support classes and various constants.

Classes

LabelType: label type. LinkType: link type. ImageExt: image extention.

DataType

Bases: Enum

DataType Enumerator.

Source code in causalflow/basics/constants.py 
34
35
36
37
38
class DataType(Enum):
    """DataType Enumerator."""

    Continuos = 0
    Discrete = 1

ImageExt

Bases: Enum

ImageExt Enumerator.

Source code in causalflow/basics/constants.py 
50
51
52
53
54
55
class ImageExt(Enum):
    """ImageExt Enumerator."""

    PNG = ".png"
    PDF = ".pdf"
    JPG = ".jpg"

LabelType

Bases: Enum

LabelType Enumerator.

Source code in causalflow/basics/constants.py 
26
27
28
29
30
31
class LabelType(Enum):
    """LabelType Enumerator."""

    Lag = "Lag"
    Score = "Score"
    NoLabels = "NoLabels"

LinkType

Bases: Enum

LinkType Enumerator.

Source code in causalflow/basics/constants.py 
41
42
43
44
45
46
47
class LinkType(Enum):
    """LinkType Enumerator."""

    Directed = "-->"
    Uncertain = "o-o"
    Bidirected = "<->"
    HalfUncertain = "o->"

This module provides the Logger class.

Classes

Logger: class responsible for the log.

Logger

Bases: object

Logger class.

Source code in causalflow/basics/logger.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
42
43
44
45
46
47
class Logger(object):
    """Logger class."""

    def __init__(self, path, clean_console = True):
        """
        Class constructor.

        Args:
            path (str): log file path.
            clean_console (bool, optional): clean console flag. Defaults to True.
        """
        self.terminal = sys.stdout
        self.log = open(path, "w")
        if clean_console: cls()

    def write(self, message):
        """
        Write message.

        Args:
            message (str): log msg.
        """
        self.terminal.write(message)
        self.log.write(message)

    def flush(self):
        """python3 compatibility."""
        # this flush method is needed for python 3 compatibility.
        # this handles the flush command by doing nothing.
        # you might want to specify some extra behavior here.
        pass

    def close(self):
        """Close logger."""
        sys.stdout = sys.__stdout__
        self.log.close()

__init__(path, clean_console=True)

Class constructor.

Parameters:

Name Type Description Default
path str

log file path.

 required
clean_console bool

clean console flag. Defaults to True.

 True
Source code in causalflow/basics/logger.py 
15
16
17
18
19
20
21
22
23
24
25
def __init__(self, path, clean_console = True):
    """
    Class constructor.

    Args:
        path (str): log file path.
        clean_console (bool, optional): clean console flag. Defaults to True.
    """
    self.terminal = sys.stdout
    self.log = open(path, "w")
    if clean_console: cls()

close()

Close logger.

Source code in causalflow/basics/logger.py 
44
45
46
47
def close(self):
    """Close logger."""
    sys.stdout = sys.__stdout__
    self.log.close()

flush()

python3 compatibility.

Source code in causalflow/basics/logger.py 
37
38
39
40
41
42
def flush(self):
    """python3 compatibility."""
    # this flush method is needed for python 3 compatibility.
    # this handles the flush command by doing nothing.
    # you might want to specify some extra behavior here.
    pass

write(message)

Write message.

Parameters:

Name Type Description Default
message str

log msg.

 required
Source code in causalflow/basics/logger.py 
27
28
29
30
31
32
33
34
35
def write(self, message):
    """
    Write message.

    Args:
        message (str): log msg.
    """
    self.terminal.write(message)
    self.log.write(message)

This module provides methods for computing various metrics.

FNR(gt, cm, alsoOrient=False)

Compute False Negative Rate between ground-truth causal graph and the estimated one.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
float float

false negative rate.
Source code in causalflow/basics/metrics.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
def FNR(gt, cm, alsoOrient = False) -> float:
    """
    Compute False Negative Rate between ground-truth causal graph and the estimated one.

    Args:
        gt (dict): ground-truth SCM.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        float: false negative rate.
    """
    fn = get_FN(gt, cm, alsoOrient)
    tp = get_TP(gt, cm, alsoOrient)
    if tp + fn == 0: return 0
    return fn / (tp + fn)

FPR(gt, min_lag, max_lag, cm, alsoOrient=False)

Compute False Positve Rate between ground-truth causal graph and the estimated one.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
min_lag int

min time lag.

 required
max_lag int

max time lag.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
float float

false positive rate.
Source code in causalflow/basics/metrics.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def FPR(gt, min_lag, max_lag, cm, alsoOrient = False) -> float:
    """
    Compute False Positve Rate between ground-truth causal graph and the estimated one.

    Args:
        gt (dict): ground-truth SCM.
        min_lag (int): min time lag.
        max_lag (int): max time lag.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        float: false positive rate.
    """
    fp = get_FP(gt, cm, alsoOrient)
    tn = get_TN(gt, min_lag, max_lag, cm, alsoOrient)
    if tn + fp == 0: return 0
    return fp / (tn + fp)

TNR(gt, min_lag, max_lag, cm, alsoOrient=False)

Compute True Negative Rate between ground-truth causal graph and the estimated one.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
min_lag int

min time lag.

 required
max_lag int

max time lag.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
float float

true negative rate.
Source code in causalflow/basics/metrics.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def TNR(gt, min_lag, max_lag, cm, alsoOrient = False) -> float:
    """
    Compute True Negative Rate between ground-truth causal graph and the estimated one.

    Args:
        gt (dict): ground-truth SCM.
        min_lag (int): min time lag.
        max_lag (int): max time lag.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        float: true negative rate.
    """
    tn = get_TN(gt, min_lag, max_lag, cm, alsoOrient)
    fp = get_FP(gt, cm, alsoOrient)
    if tn + fp == 0: return 0
    return tn / (tn + fp)

TPR(gt, cm, alsoOrient=False)

Compute True Positive Rate between ground-truth causal graph and the estimated one.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
float float

true positive rate.
Source code in causalflow/basics/metrics.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def TPR(gt, cm, alsoOrient = False) -> float:
    """
    Compute True Positive Rate between ground-truth causal graph and the estimated one.

    Args:
        gt (dict): ground-truth SCM.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        float: true positive rate.
    """
    tp = get_TP(gt, cm, alsoOrient)
    fn = get_FN(gt, cm, alsoOrient)
    if tp + fn == 0: return 0
    return tp / (tp + fn)

f1_score(gt, cm, alsoOrient=False)

Compute F1-score between ground-truth causal graph and the estimated one.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
float float

F1-score.
Source code in causalflow/basics/metrics.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def f1_score(gt, cm, alsoOrient = False) -> float:
    """
    Compute F1-score between ground-truth causal graph and the estimated one.

    Args:
        gt (dict): ground-truth SCM.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        float: F1-score.
    """
    p = precision(gt, cm, alsoOrient)
    r = recall(gt, cm, alsoOrient)
    if p + r == 0: return 0
    return (2 * p * r) / (p + r)

fully_connected_dag(features, min_lag, max_lag, alsoOrient=False)

Build a fully connected DAG.

Parameters:

Name Type Description Default
features list

variable list

 required
min_lag int

min time lag.

 required
max_lag int

max time lag.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
dict dict

fully connected dag
Source code in causalflow/basics/metrics.py 
 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
def fully_connected_dag(features, min_lag, max_lag, alsoOrient = False) -> dict:
    """
    Build a fully connected DAG.

    Args:
        features (list): variable list
        min_lag (int): min time lag.
        max_lag (int): max time lag.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        dict: fully connected dag
    """
    if not alsoOrient:
        g = {f: list() for f in features}
        for t in g:
            for s in g:
                for l in range(min_lag, max_lag + 1):
                    if s == t and l == 0: continue 
                    g[t].append((s, -abs(l)))
    else:
        g = {f: {} for f in features}
        for t in g:
            for s in g:
                for l in range(min_lag, max_lag + 1):
                    if s == t and l == 0: continue 
                    g[t][(s, -abs(l))] = ['-->', 'o->', '<->', 'o-o']
    return g

get_FN(gt, cm, alsoOrient=False)

Compute false negative number: edge present in the groundtruth but absent in the causal model.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
int int

false negative.
Source code in causalflow/basics/metrics.py 
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
def get_FN(gt, cm, alsoOrient = False) -> int:
    """
    Compute false negative number: edge present in the groundtruth but absent in the causal model.

    Args:
        gt (dict): ground-truth SCM.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        int: false negative.
    """
    counter = 0
    if not alsoOrient:
        for t in gt.keys():
            for s in gt[t]:
                if s not in cm[t]: counter += 1
    else:
        for t in gt.keys():
            for s in gt[t]:
                if s not in cm[t]: counter += 3
                else:
                    if gt[t][s][0] != cm[t][s][0]: counter += 1
                    if gt[t][s][1] != cm[t][s][1]: counter += 1
                    if gt[t][s][2] != cm[t][s][2]: counter += 1
    return counter

get_FP(gt, cm, alsoOrient=False)

Compute false positive number: edge present in the causal model but absent in the groundtruth.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
int int

false positive.
Source code in causalflow/basics/metrics.py 
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
def get_FP(gt, cm, alsoOrient = False) -> int:
    """
    Compute false positive number: edge present in the causal model but absent in the groundtruth.

    Args:
        gt (dict): ground-truth SCM.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        int: false positive.
    """
    counter = 0
    if not alsoOrient:
        for t in cm.keys():
            for s in cm[t]:
                if s not in gt[t]: counter += 1
    else:
        for t in cm.keys():
            for s in cm[t]:
                if s not in gt[t]: counter += 3
                else:
                    if gt[t][s][0] != cm[t][s][0]: counter += 1
                    if gt[t][s][1] != cm[t][s][1]: counter += 1
                    if gt[t][s][2] != cm[t][s][2]: counter += 1
    return counter

get_TN(gt, min_lag, max_lag, cm, alsoOrient=False)

Compute true negative number: edge absent in the groundtruth and absent in the causal model.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
min_lag int

min time lag.

 required
max_lag int

max time lag.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
int int

true negative.
Source code in causalflow/basics/metrics.py 
 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
def get_TN(gt, min_lag, max_lag, cm, alsoOrient = False) -> int:
    """
    Compute true negative number: edge absent in the groundtruth and absent in the causal model.

    Args:
        gt (dict): ground-truth SCM.
        min_lag (int): min time lag.
        max_lag (int): max time lag.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        int: true negative.
    """
    fullg = fully_connected_dag(list(gt.keys()), min_lag, max_lag, alsoOrient)
    counter = 0
    if not alsoOrient:
        gt_TN = deepcopy(fullg)

        # Build the True Negative graph [complementary graph of the ground-truth]
        for t in fullg:
            for s in fullg[t]:
                if s in gt[t]:
                    gt_TN[t].remove(s)

        for t in gt_TN.keys():
            for s in gt_TN[t]:
                if s not in cm[t]: counter += 1
    else:
        gt_TN = deepcopy(fullg)

        # Build the True Negative graph [complementary graph of the ground-truth]
        for t in fullg:
            for s in fullg[t]:
                for edge in fullg[t][s]:
                    if s not in gt[t]: continue
                    if edge == gt[t][s]:
                        gt_TN[t][s].remove(edge)
                if len(fullg[t][s]) == 0: del gt_TN[t][s]

        for t in gt_TN.keys():
            for s in gt_TN[t]:
                if s not in cm[t]: 
                    counter += len(gt_TN[t][s])
                else:
                    for edge in gt_TN[t][s]:
                        if edge != cm[t][s]: counter += 1

    return counter

get_TP(gt, cm, alsoOrient=False)

Compute true positive number: edge present in the causal model and present in the groundtruth.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
int int

true positive.
Source code in causalflow/basics/metrics.py 
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
def get_TP(gt, cm, alsoOrient = False) -> int:
    """
    Compute true positive number: edge present in the causal model and present in the groundtruth.

    Args:
        gt (dict): ground-truth SCM.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        int: true positive.
    """
    counter = 0
    if not alsoOrient:
        for t in cm.keys():
            for s in cm[t]:
                if s in gt[t]: counter += 1
    else:
        for t in cm.keys():
            for s in cm[t]:
                if s in gt[t]:
                    if gt[t][s][0] == cm[t][s][0]: counter += 1
                    if gt[t][s][1] == cm[t][s][1]: counter += 1
                    if gt[t][s][2] == cm[t][s][2]: counter += 1
    return counter

precision(gt, cm, alsoOrient=False)

Compute Precision between ground-truth causal graph and the estimated one.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
float float

precision.
Source code in causalflow/basics/metrics.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def precision(gt, cm, alsoOrient = False) -> float:
    """
    Compute Precision between ground-truth causal graph and the estimated one.

    Args:
        gt (dict): ground-truth SCM.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        float: precision.
    """
    tp = get_TP(gt, cm, alsoOrient)
    fp = get_FP(gt, cm, alsoOrient)
    if tp + fp == 0: return 0
    return tp/(tp + fp)

recall(gt, cm, alsoOrient=False)

Compute Recall between ground-truth causal graph and the estimated one.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
float float

recall.
Source code in causalflow/basics/metrics.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def recall(gt, cm, alsoOrient = False) -> float:
    """
    Compute Recall between ground-truth causal graph and the estimated one.

    Args:
        gt (dict): ground-truth SCM.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        float: recall.
    """
    tp = get_TP(gt, cm, alsoOrient)
    fn = get_FN(gt, cm, alsoOrient)
    if tp + fn == 0: return 0
    return tp/(tp + fn)

shd(gt, cm, alsoOrient=False)

Compute Structural Hamming Distance between ground-truth causal graph and the estimated one.

Parameters:

Name Type Description Default
gt dict

ground-truth SCM.

 required
cm dict

estimated SCM.

 required
alsoOrient bool

Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

 False

Returns:

Name Type Description
int int

shd.
Source code in causalflow/basics/metrics.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def shd(gt, cm, alsoOrient = False) -> int:
    """
    Compute Structural Hamming Distance between ground-truth causal graph and the estimated one.

    Args:
        gt (dict): ground-truth SCM.
        cm (dict): estimated SCM.
        alsoOrient (bool, optional): Flag to compute the metric considering also the orientation of the link and not only its presence. Default False.

    Returns:
        int: shd.
    """
    fn = get_FN(gt, cm, alsoOrient)
    fp = get_FP(gt, cm, alsoOrient)
    return fn + fp

This module provides utilities methods.

cls()

Clear terminal.

Source code in causalflow/basics/utils.py 
 9
10
11
def cls():
    """Clear terminal."""
    os.system('cls' if os.name == 'nt' else 'clear')

create_results_folder(resfolder)

Create results folder if doesn't exist.

Parameters:

Name Type Description Default
resfolder str

result folder's name.

 required
Source code in causalflow/basics/utils.py 
28
29
30
31
32
33
34
35
def create_results_folder(resfolder):
    """
    Create results folder if doesn't exist.

    Args:
        resfolder (str): result folder's name.
    """
    Path(resfolder).mkdir(parents=True, exist_ok=True)

get_selectorpath(resfolder)

Return log file path.

Parameters:

Name Type Description Default
resfolder str

result folder.

 required

Returns:

Type Description
str

log file path.
Source code in causalflow/basics/utils.py 
14
15
16
17
18
19
20
21
22
23
24
25
def get_selectorpath(resfolder):
    """
    Return log file path.

    Args:
        resfolder (str): result folder.

    Returns:
        (str): log file path.
    """
    Path(resfolder).mkdir(parents=True, exist_ok=True)
    return SEP.join([resfolder, LOG_FILENAME]), SEP.join([resfolder, RES_FILENAME]), SEP.join([resfolder, DAG_FILENAME]), SEP.join([resfolder, TSDAG_FILENAME])

remove_from_list(list, item)

Create a copy of a list and remove from it an item.

Parameters:

Name Type Description Default
list list

list.

 required
item any

item to remove.

 required

Returns:

Name Type Description
list list()

new list without the item.
Source code in causalflow/basics/utils.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def remove_from_list(list: list(), item) -> list():
    """
    Create a copy of a list and remove from it an item.

    Args:
        list (list): list.
        item (any): item to remove.

    Returns:
        list: new list without the item.
    """
    tmp = copy.deepcopy(list)
    tmp.remove(item)
    return tmp