#!/usr/bin/python3 -ttOO ''' Artificial Intelligence Research Group University of Lleida ''' # Libraries import sys import math import ast import xml.etree.ElementTree as ET import networkx import random import time import psutil import pandas import numpy import scipy import reddit_at # Global functions def sentiment(sentiment_distribution): ''' Computes the sentiment from a sentiment distribution of 5 values --> [-2, 2] ''' sentiment_relevance = [-2, -1, 0, 1, 2] res = [a * b for a, b in zip(sentiment_relevance, sentiment_distribution)] return sum(res) def scale_weight(weight, args): ''' Scales the weight using a log function ''' if weight >= 1: return int(math.floor(math.log(weight, args.log_base)) + 1) else: return 0 def get_weighted_color(base_color, min_weight, max_weight, w, hw = 0xCF): #hw = 0xFF #0xCF # Highest light color to not be completely white if w >= max_weight: hw = 0 elif max_weight > min_weight: hw = int(hw * (float(max_weight - w) / float(max_weight - min_weight))) color = [a | b for a, b in zip(base_color, [hw, hw, hw])] contrast = '#FFFFFF' if hw < 0x80 else '#000000' return contrast, color def ecai2023_bounded_normal_distribution(mean, minv, maxv ): ''' Bounded normal distribution, like normal distribution but the two values are always in [min, max] ''' mu = mean # sigma = 2 / 3 sigma = 2.0 / (3.0 + (mean * 10.0)) if (mean > 0.0) else 2.0 / (3.0 + (-mean * 10.0)) f = scipy.stats.truncnorm((minv - mu) / sigma, (maxv - mu) / sigma, loc = mu, scale = sigma) return f.rvs() def ecai2023_getTNormal(minv, maxv, sigma, mu): print(" Params for one-sided tnormal: ", minv, maxv, sigma, mu) return scipy.stats.truncnorm((minv - mu) / sigma, (maxv - mu) / sigma, loc = mu, scale = sigma) # Classes class UDebG(): ''' User-based Debate Graph ''' def __init__(self, DebT, root_id, args): self.DebT = DebT self.DebT_root_id = root_id self.UDebG = networkx.DiGraph() self.max_weight = None self.min_weight = None self.max_pos_edge_weight = None self.min_pos_edge_weight = None self.max_neg_edge_weight = None self.min_neg_edge_weight = None self.VAF_accepted = None if args.user == 'wia2021' or not args.input_file: random.seed(args.seed) if args.seed is not None: numpy.random.seed(int(args.seed)) # For scipy if not args.input_file: self.ecai2023_gen_UDebG(args) else: self.wia2021_set_nodes() self.wia2021_set_edges() self.wia2021_UDebG(args) if args.algorithm.startswith('g'): if len(args.algorithm) == 2: self.wia2021_polarized_partition_greedy(args) elif len(args.algorithm) == 4: self.ccia2022_polarized_partition_greedy(args) else: sys.exit('ERROR: Number of algorithm parameters (%s) not recognized.' % args.algorithm) else: sys.exit('ERROR: Algorithm (%s) not recognized.' % args.algorithm) elif args.user == 'mdai2020': self.mdai2020_set_nodes() self.mdai2020_set_edges() self.mdai2020_draw_DebT(args) print('MDAI 2020 UDebG...') self.mdai2020_UDebG() self.UDebG.remove_node('0') # No root node for the solution self.mdai2020_skeptical() self.mdai2020_VAF_valuation(args) if args.draw_graphs: self.mdai2020_draw_UDebG(args) self.mdai2020_UDebG_to_xml(args) self.VAF_accepted = reddit_at.VAF_solver(args, "%s.udebg.xml" % args.input_file) if args.draw_graphs: self.mdai2020_draw_UDebG(args) def ecai2023_gen_UDebG(self, args): ''' Generates a random UDebG with specific properties arg.params comma separated format: N,B - N: number of nodes (int) - B: beta parameter for p value (float) ''' print('ecai2023 Generating random UDebG...') try: N, B, DEGCTE = args.params.split(',') N = int(N) B = float(B) DEGCTE = float(DEGCTE) except: sys.exit('ERROR: Incorrect parameters for UDebG random generation (%s)' % (args.params)) A = args.alpha # alpha parameter for nodes if N < 0 or A > 1 or A < 0 or B > 1 or B < 0 or args.log_base < 2 or DEGCTE < 1.0: sys.exit('ERROR: Incorrect parameter ({}-{}-{}-{}-{})'.format(N, A, B, DEGCTE, args.log_base)) E = int(math.ceil(math.log(N, args.log_base))*DEGCTE) # maximum number of edges print(' Parameters: nodes = {}, max out edges per node = {}, alpha = {}, beta = {}, DegMultConstant = {}, seed = {}'.format(N, E, A, B, DEGCTE, args.seed)) args.input_file = 'rnd-UDebG-{}-{}-{}-{}-{}-{}'.format(N, E, A, B, args.log_base, args.seed) self.ecai2023_set_nodes(N, A) self.ecai2023_set_edges(E, B) if args.draw_graphs: self.wia2021_draw_UDebG(args) if args.scip_output: self.ccia2022_scip_output(args) def ecai2023_set_nodes(self, N, A): ''' Generate N nodes for the random UDebG with opinion weighting scheme in [-A, A] ''' fl = ecai2023_getTNormal(-A, 0, 1.0 / (1.0 + (A * 20.0)), -A) fr = ecai2023_getTNormal(0, A, 1.0 / (1.0 + (A * 20.0)), A) for node_id in range(1, N + 1): # No root node with Id = 0 if (A == 0.0): ows = random.uniform(-A, A) else: ows = fr.rvs() if (random.randint(0, 1)) else fl.rvs() self.UDebG.add_node('user%i' % node_id, opinion_ws = ows, node_id = node_id) def ecai2023_set_edges(self, E, B ): ''' Generate a maximum of E out edges for each node for the random UDebG Parameter B is ignored ''' for n1 in self.UDebG.nodes(): list_nodes = list(self.UDebG.nodes()) list_nodes.remove(n1) # No self-answers for e in range(random.randint(1, E)): n2 = list_nodes.pop(random.randint(0, len(list_nodes) - 1)) n1_ows = self.UDebG.nodes[n1]['opinion_ws'] n2_ows = self.UDebG.nodes[n2]['opinion_ws'] # when they are in different sides: if ( (n1_ows * n2_ows < 0.0) or (n1_ows == 0.0 and n2_ows > 0.0) or (n1_ows > 0.0 and n2_ows == 0.0) ): dist = -abs(n1_ows - n2_ows) * abs(n1_ows) else: # in the same side dist = 2 * abs(n1_ows) - abs(n1_ows - n2_ows) #dist = abs(n1_ows - n2_ows) * abs(n1_ows) # Maximum range of [0, 2] for alpha 1 #if n2_ows == 0: # dist must be sign of n1_ows # dist = dist if n1_ows > 0 else -dist #else: # if n1_ows * n2_ows < 0: # Different sign, opposite opinions # dist = -dist #if dist > 0: # Max range of dist in [0, 1] # dist = dist * 2 p = 1 # NOTE: set to 1 (or 0) in order to w = iws, but intended to be random.uniform(0, B) w = ecai2023_bounded_normal_distribution(dist, -2, 2) iws = (p, w) # ([0, 1], [-2, 2]) self.UDebG.add_edge(n1, n2, interaction_ws = iws) def wia2021_set_nodes(self): ''' UDebG has a node for each user and stores each comment of the user in a list as node data ''' node_id = 1 for n, nd in self.DebT.nodes(data = True): if 'title' in nd['data'].attrib: # Root comment self.UDebG.add_node('0', data = nd, node_id = 0) # Id = 0 for root node (special node) else: # Regular comment user = nd['data'].get('author') if user not in self.UDebG: self.UDebG.add_node(user, data = [nd], node_id = node_id) node_id = node_id + 1 else: self.UDebG.nodes[user]['data'].append(nd) def wia2021_set_edges(self): ''' UDebG has an edge between u_a and u_b if there is a reply form u_a to u_b ''' for e1, e2, ed in self.DebT.edges(data = True): # u_e1 replies to u_e2 u_e1 = self.DebT.nodes[e1]['data'].get('author') if 'title' in self.DebT.nodes[e2]['data'].attrib: # Reply to root comment u_e2 = '0' else: u_e2 = self.DebT.nodes[e2]['data'].get('author') if not self.UDebG.has_edge(u_e1, u_e2): self.UDebG.add_edge(u_e1, u_e2, data = [ed]) else: self.UDebG[u_e1][u_e2]['data'].append(ed) def wia2021_UDebG(self, args): ''' WIA2021 UDebG. Name of variables following paper notation ''' print('Generating UDebG for wia2021...') # Discard auto-replies for user in self.UDebG.nodes(): if self.UDebG.has_edge(user, user): self.UDebG.remove_edge(user, user) # No root node for the solution self.UDebG.remove_node('0') # Opinion weighting scheme for nodes (users) print(' Number of nodes UDebG = {}'.format(self.UDebG.number_of_nodes())) for n, nd in self.UDebG.nodes(data = True): s = 0 for cs in nd['data']: s = s + cs['side'] s = s / len(nd['data']) nd['opinion_ws'] = s if s < -1 or s > 1: sys.exit('ERROR: S value (%f) out of range [-1, 1].' % s) # Interaction weighting scheme for edges for e1, e2, ed in self.UDebG.edges(data = True): p = 0 w = 0 for d in ed['data']: cu1 = d['data'].find('t').get('id') W = self.DebT.nodes[cu1]['sentiment_not_normalized'] if W > 0: p = p + 1 # Counts edges with positive W w = w + W # Sums all W p = p / len(ed['data']) w = w / len(ed['data']) # Aggregation operator (mean) ed['interaction_ws'] = (p, w) if p < 0 or p > 1 or w < -2 or w > 2: sys.exit('ERROR: pair (p, w) (%f, %f) out of range ([0, 1], [-2, 2]).' % (p, w)) if args.draw_graphs: self.wia2021_draw_UDebG(args) if args.scip_output: self.ccia2022_scip_output(args) sys.exit() def wia2021_sideness_consistency(self, L, R): ''' WIA2021 We define the sideness consistency of two sides L and R as: SC(L, R, G) = LC(L, G) * RC (R, G) ''' LC = 0 RC = 0 for n in L: S = self.UDebG.nodes[n]['opinion_ws'] if S <= 0: LC = LC - S for n in R: S = self.UDebG.nodes[n]['opinion_ws'] if S > 0: RC = RC + S cardC = len(L) + len(R) LC = LC / cardC RC = RC / cardC SC = LC * RC if SC < 0 or SC > 0.25: sys.exit('ERROR: SC (%f) out of range [0, 0.25].' % SC) return LC, RC, SC def wia2021_interactions_sentiment(self, L, R): ''' WIA2021 We define the sentiment of the interactions between users of different sides as follows... ''' sum_inter = 0 for e1, e2, ed in self.UDebG.edges(data = True): p = ed['interaction_ws'][0] w = ed['interaction_ws'][1] if (e1 in L and e2 in R) or (e1 in R and e2 in L): sum_inter = sum_inter + (-(2 * ((p - 0.5) ** 2) + 0.5)) * w SWeight = sum_inter / self.UDebG.number_of_edges() + 2 if SWeight < 0 or SWeight > 4: sys.exit('ERROR: SWeight (%f) out of range [0, 4].' % SWeight) return SWeight def wia2021_BipPol(self, L, R, all_stats = False): ''' WIA2021 Combine both measures SC and SWeight to define the Bipartition Polarization level of a given partition (L, R) ''' LC, RC, SC = self.wia2021_sideness_consistency(L, R) SWeight = self.wia2021_interactions_sentiment(L, R) BipPol = SC * SWeight if all_stats: return LC, RC, SC, SWeight, BipPol else: return BipPol def wia2021_initial_partition(self, args): ''' WIA2021 Initial partition for polarized partition algorithm See docstring of wia2021_polarized_partition_greedy for more information ''' L = [] R = [] if len(args.algorithm) > 1 and args.algorithm[1] == '0': # Random initialization for n, nd in self.UDebG.nodes(data = True): if random.random() < 0.5: L.append(n) else: R.append(n) elif len(args.algorithm) > 1 and args.algorithm[1] == '1': # Negatives and neutral to L and positives to R for n, nd in self.UDebG.nodes(data = True): if nd['opinion_ws'] <= 0: L.append(n) else: R.append(n) elif len(args.algorithm) > 1 and args.algorithm[1] == '2': # L with P = (1 − S(c)) / 2 and in R with 1 − P for n, nd in self.UDebG.nodes(data = True): P = (1 - nd['opinion_ws']) / 2 if random.random() < P: L.append(n) else: R.append(n) else: sys.exit('ERROR: Algorithm settings (%s) with no partition initialization.' % args.algorithm) return L, R def wia2021_find_better_v(self, L, R, LtoR): ''' WIA2021 Find a node in L such that removing it from L and adding it to R increases BipPol value. If LtoR is false, finds a node in R such that removing it from R and adding it to L increases BipPol value. ''' L1 = list(L) R1 = list(R) BipPol = self.wia2021_BipPol(L, R) if LtoR: # Search for a node from L to move to R for n in L: L1.remove(n) R1.append(n) if self.wia2021_BipPol(L1, R1) > BipPol: return n L1.append(n) R1.pop() else: # Search for a node from R to move to L for n in R: R1.remove(n) L1.append(n) if self.wia2021_BipPol(L1, R1) > BipPol: return n R1.append(n) L1.pop() return None def wia2021_polarized_partition_greedy(self, args): ''' WIA2021 Polarized Partition greedy algorithm. Set L contains mainly users in the negative side of the debate and the set R contains mainly user in the positive side of the debate. Algorithm parameters args.algorithm: - First char of the string, 'g': Chooses this Greedy Polarized Partition algorithm - Second char of the string, sets initial partition: - '0': Distributes uniformly at random in either L or R - '1': Order vertices by polarity assigning the negatives and neutral to L and the positives to R - '2': Place user's opinion c randomly in L with probability P = (1 − S(c)) / 2 and in R with probability 1 − P ''' print('WIA2021 Greedy Bipartite Polarization algorithm...') L, R = self.wia2021_initial_partition(args) improving = True steps = 0 init_BipPol = self.wia2021_BipPol(L, R) print(" Initial BipPol = %f" % init_BipPol) while improving and steps < len(self.UDebG): print('\r Step %i/%i...' % (steps + 1, len(self.UDebG)), end = '') improving = False v = self.wia2021_find_better_v(L, R, LtoR = True) if v: L.remove(v) R.append(v) improving = True v = self.wia2021_find_better_v(L, R, LtoR = False) if v: R.remove(v) L.append(v) improving = True steps = steps + 1 if not improving: print(' not improving...', end = '') LC, RC, SC, SWeight, BipPol = self.wia2021_BipPol(L, R, all_stats = True) print("\n Final BipPol = %f" % BipPol) if args.draw_graphs: self.wia2021_draw_UDebG(args, L) self.wia2021_stats_to_file(args, init_BipPol, steps, LC, RC, SC, SWeight, BipPol, L, R) def ccia2022_select_neighbor_HC(self, L, R): ''' CCIA2022 Select Neighbor with Hill Climbing strategy and applies changes: Find a node in L such that removing it from L and adding it to R increases BipPol value, then do the same from R to L. ''' BipPol = self.wia2021_BipPol(L, R) changes = 0 i = 0 while i < len(L): n = L.pop(0) R.append(n) bp = self.wia2021_BipPol(L, R) if bp > BipPol: BipPol = bp changes = changes + 1 break L.append(n) R.pop() i = i + 1 i = 0 while i < len(R): n = R.pop(0) L.append(n) bp = self.wia2021_BipPol(L, R) if bp > BipPol: changes = changes + 1 break R.append(n) L.pop() i = i + 1 return changes def ccia2022_select_neighbor_SAHC(self, L, R): ''' CCIA2022 Select Neighbor with Steepest Ascent Hill Climbing strategy and applies changes: Find a node that changing it from L to R or from R to L increases more the BipPol value. Ties broken randomly. ''' best_bp = self.wia2021_BipPol(L, R) best_init_bp = best_bp imp_nbs = [] i = 0 while i < len(L): n = L.pop(i) R.append(n) bp = self.wia2021_BipPol(L, R) if bp >= best_bp and bp > best_init_bp: if bp > best_bp: best_bp = bp imp_nbs = [] imp_nbs.append((i, L, R)) L.insert(i, n) R.pop() i = i + 1 i = 0 while i < len(R): n = R.pop(i) L.append(n) bp = self.wia2021_BipPol(L, R) if bp >= best_bp and bp > best_init_bp: if bp > best_bp: best_bp = bp imp_nbs = [] imp_nbs.append((i, R, L)) R.insert(i, n) L.pop() i = i + 1 if len(imp_nbs) == 0: return 0 elif len(imp_nbs) > 0: i, from_set, to_set = random.choice(imp_nbs) n = from_set.pop(i) to_set.append(n) return 1 def ccia2022_select_neighbor_funtion(self, args): ''' CCIA2022 Select "select_neighbor" function according to third char of algorithm parameters ''' if args.algorithm[2] == '0': return self.ccia2022_select_neighbor_HC elif args.algorithm[2] == '1': return self.ccia2022_select_neighbor_SAHC else: sys.exit('ERROR: Algorithm parameter for select_neighbor function (%s) not recognized.' % args.algorithm) def ccia2022_select_restarts(self, args): ''' CCIA2022 Select number of restarts according to fourth char of algorithm parameters ''' if args.algorithm[3] == '0': return 1 elif args.algorithm[3] == '1': return 10 else: sys.exit('ERROR: Algorithm parameter for select_restarts function (%s) not recognized.' % args.algorithm) def ccia2022_randomize_sets(self, L, R, noise): ''' CCIA2022 Randomize sets by noise percentage (probability to switch set) ''' R_init_size = len(R) i = len(L) - 1 while i >= 0: if random.random() < noise: n = L.pop(i) R.append(n) i = i - 1 i = R_init_size - 1 while i >= 0: if random.random() < noise: n = R.pop(i) L.append(n) i = i - 1 def ccia2022_polarized_partition_greedy(self, args): ''' CCIA2022 Polarized Partition greedy algorithm. Set L contains mainly users in the negative side of the debate and the set R contains mainly user in the positive side of the debate. Algorithm parameters args.algorithm: - First char of the string, 'g': Chooses this Greedy Polarized Partition algorithm - Second char of the string sets initial partition: - '0': Distributes uniformly at random in either L or R - '1': Order vertices by polarity assigning the negatives and neutral to L and the positives to R - '2': Place user's opinion c randomly in L with probability P = (1 − S(c)) / 2 and in R with probability 1 − P - Third char of the string sets better solution strategy: - '0': Hill Climbing strategy, pick the first neighbor that improve - '1': Steepest Ascent Hill Climbing strategy, pick the neighbor that improves more - Fourth char of the string sets the restarts - '0': No restarts - '1': 10 restarts ''' print('CCIA2022 Greedy Bipartite Polarization algorithm...') p = psutil.Process() init_cpu_time = p.cpu_times()[0] select_neighbor = self.ccia2022_select_neighbor_funtion(args) L, R = self.wia2021_initial_partition(args) restarts = self.ccia2022_select_restarts(args) init_BipPol = self.wia2021_BipPol(L, R) best_BipPol = init_BipPol best_L = L[:] best_R = R[:] print(" Initial BipPol = %f" % init_BipPol) while restarts: changes = 1 steps = 0 while changes and steps < len(self.UDebG): print('\r Step %i/%i...' % (steps + 1, len(self.UDebG)), end = '') changes = select_neighbor(L, R) steps = steps + changes if not changes: print(' not improving...') restarts = restarts - 1 BipPol = self.wia2021_BipPol(L, R) print(" Try final BipPol = %f" % BipPol) if best_BipPol < BipPol: best_BipPol = BipPol best_L = L[:] best_R = R[:] if restarts: self.ccia2022_randomize_sets(L, R, 0.1) BipPol = self.wia2021_BipPol(L, R) print(" Try init BipPol = %f" % BipPol) LC, RC, SC, SWeight, BipPol = self.wia2021_BipPol(best_L, best_R, all_stats = True) print(" Final best BipPol = %f" % BipPol) final_cpu_time = p.cpu_times()[0] print(" CPU time = %0.3f" % (final_cpu_time - init_cpu_time)) if args.draw_graphs: self.wia2021_draw_UDebG(args, best_L) self.wia2021_stats_to_file(args, init_BipPol, steps, LC, RC, SC, SWeight, BipPol, best_L, best_R) def ccia2022_scip_output(self, args): ''' UDebG Bipartition problem output format for SCIP solver ''' print('Writing UDebG Bipartition problem in SCIP format...') file_vertices = '%s.udebg-vertices.scp' % args.input_file file_edges = '%s.udebg-edges.scp' % args.input_file file_stats = '%s.stats' % args.input_file # Vertices values = [] with open(file_vertices, 'w') as f: for n, nd in self.UDebG.nodes(data = True): f.write('v{} {}\n'.format(self.UDebG.nodes[n]['node_id'], nd['opinion_ws'])) values.append(nd['opinion_ws']) df = pandas.DataFrame(data = values, columns = ['Node weights']) stats = str(df.describe()) + '\n' # Edges values = [] with open(file_edges, 'w') as f: for e1, e2, ed in self.UDebG.edges(data = True): p = ed['interaction_ws'][0] w = ed['interaction_ws'][1] cost = (-(2 * ((p - 0.5) ** 2) + 0.5)) * w f.write('v{} v{} {}\n'.format(self.UDebG.nodes[e1]['node_id'], self.UDebG.nodes[e2]['node_id'], cost)) values.append(cost) df = pandas.DataFrame(data = values, columns = ['Edge weights']) stats += str(df.describe()) + '\n' # Compute stats for the out degree of each node od = [t[1] for t in self.UDebG.out_degree(self.UDebG.nodes())] stats += f'Min out degree: {min(od)}\n' stats += f'Max out degree: {max(od)}\n' stats += f'Mean out degree: {numpy.mean(od)}\n' # Write stats to file with open(file_stats, 'w') as f: f.write(stats) def wia2021_stats_to_file(self, args, init_BipPol, steps, LC, RC, SC, SWeight, BipPol, L, R): ''' Write Greedy polarized partition algorithm statistics and information to file ''' print('Writing statistics to file...') out_str = 'WIA2021 stats\n------------\n' out_str += 'Timestamp = %s\n' % time.ctime() out_str += 'Input file = %s\n' % args.input_file out_str += 'UDebG #nodes = %i\n' % self.UDebG.number_of_nodes() out_str += 'UDebG #edges = %i\n' % self.UDebG.number_of_edges() out_str += 'UDebG ratio #edges/#nodes = %f\n' % (self.UDebG.number_of_edges() / self.UDebG.number_of_nodes()) out_str += 'Algorithm = %s\n' % args.algorithm out_str += 'Initial BipPol = %f\n' % init_BipPol out_str += 'Algorithm steps = %i\n' % steps out_str += 'Final BipPol = %f\n' % BipPol out_str += 'LC = %f\n' % LC out_str += 'RC = %f\n' % RC out_str += 'SC = %f\n' % SC out_str += 'SWeight = %f\n' % SWeight out_str += '#users in L = %i\n' % len(L) out_str += '#users in R = %i\n' % len(R) out_str += 'Set L = %s\n' % str([self.UDebG.nodes[n]['node_id'] for n in L]) out_str += 'Set R = %s\n' % str([self.UDebG.nodes[n]['node_id'] for n in R]) # Write to file output_file_name = '%s.%s.info' % (args.input_file, args.algorithm) output_file = open(output_file_name, 'w') output_file.write(out_str) output_file.close() def wia2021_draw_UDebG(self, args, L = None): ''' Drawing WIA2021 UDebG ''' if L: print('Drawing wia2021 Bipartite UDebG...') output_file_name = '%s.udebg.bip.png' % args.input_file else: print('Drawing wia2021 UDebG...') output_file_name = '%s.udebg.png' % args.input_file gv = networkx.nx_agraph.to_agraph(self.UDebG) gv.node_attr['style'] = 'filled' gv.node_attr['fixedsize'] = 'true' gv.node_attr['width'] = '0.4' gv.node_attr['height'] = '0.4' for n in gv.nodes(): node_id = self.UDebG.nodes[n]['node_id'] n.attr['label'] = str(node_id) bordercolor = '#000000' n.attr['penwidth'] = 1 if L: # UDebG Bipartite if n in L: n.attr['fillcolor'] = '#FF0000' n.attr['fontcolor'] = '#FFFFFF' else: n.attr['fillcolor'] = '#0000FF' n.attr['fontcolor'] = '#FFFFFF' else: # UDebG s = self.UDebG.nodes[n]['opinion_ws'] if s > 0: # cyan = '#4FCFFF' contrast, color = get_weighted_color([0x4F, 0xCF, 0xFF], 0, 1, s, hw = 0xAF) contrast = '#000000' elif s < 0: # dark blue = '#00007F' contrast, color = get_weighted_color([0x00, 0x00, 0x7F], 0, 1, -s, hw = 0xAF) else: color = [0xFF, 0xFF, 0xFF] contrast = '#000000' n.attr['fillcolor'] = '#%s' % ''.join([c[2:].zfill(2) for c in map(hex, color)]) n.attr['fontcolor'] = contrast gv.edge_attr['color'] = '#000000' for e in gv.edges(): p = self.UDebG[e[0]][e[1]]['interaction_ws'][0] w = self.UDebG[e[0]][e[1]]['interaction_ws'][1] if w > 0: contrast, color = get_weighted_color([0x00, 0xFF, 0x00], 0, 1, p) elif w < 0: contrast, color = get_weighted_color([0xFF, 0x00, 0x00], 0, 1, 1 - p) else: color = [0x00, 0x00, 0x00] e.attr['color'] = '#%s' % ''.join([c[2:].zfill(2) for c in map(hex, color)]) gv.layout(prog = 'dot', args='-Goverlap=false -Gnodesep=0.2 -Granksep=0.2 -Grankdir=BT -GK=800 -Gstart=17 -Gmaxiter=600') gv.draw(output_file_name, format = 'png') def mdai2020_set_nodes(self): ''' UDebG has a node for each user and stores each comment of the user in a list as node data ''' # Set chronological id to comments in DebT self.DebT.nodes[self.DebT_root_id]['chrono_id'] = 0 id_list = sorted([n for n, nd in self.DebT.nodes(data = True) if 'title' not in nd['data'].attrib]) for i, c_id in enumerate(id_list): self.DebT.nodes[c_id]['chrono_id'] = i + 1 node_id = 1 for n, nd in self.DebT.nodes(data = True): if 'title' in nd['data'].attrib: # Root comment self.UDebG.add_node('0', data = nd['data'], node_id = 0) # Id = 0 for root node (special node) else: # Regular comment user = nd['data'].get('author') if user not in self.UDebG: self.UDebG.add_node(user, data = [nd['data']], node_id = node_id) node_id = node_id + 1 else: self.UDebG.nodes[user]['data'].append(nd['data']) # Print number of comments per user # print('User\tId\t#comments') # for n, nd in self.UDebG.nodes(data = True): # print('%s\t%i\t%i' % (n, nd['node_id'], len(nd['data']))) def mdai2020_set_edges(self): ''' UDebG has an edge between u_a and u_b if there is a reply form u_a to u_b ''' for e1, e2, ed in self.DebT.edges(data = True): # u_e1 replies to u_e2 u_e1 = self.DebT.nodes[e1]['data'].get('author') if 'title' in self.DebT.nodes[e2]['data'].attrib: # Reply to root comment u_e2 = '0' else: u_e2 = self.DebT.nodes[e2]['data'].get('author') if not self.UDebG.has_edge(u_e1, u_e2): self.UDebG.add_edge(u_e1, u_e2, data = [ed['data']]) else: self.UDebG[u_e1][u_e2]['data'].append(ed['data']) def mdai2020_UDebG(self): ''' MDAI 2020 requirements: Auto-replies are discarded ''' for user in self.UDebG.nodes(): if self.UDebG.has_edge(user, user): self.UDebG.remove_edge(user, user) def mdai2020_skeptical(self): ''' Skeptical sentiment weighting scheme L:E --> [-2, 2] min(sentiment) of replies from u_a to u_b if (all comments of u_a agree with u_b) or (all comments of u_a disagree with u_b) 0 otherwise ''' for u1, u2, ed in self.UDebG.edges(data = True): ed['skeptical'] = None for answer in ed['data']: cu1 = answer.find('t').get('id') s = sentiment(ast.literal_eval(self.DebT.nodes[cu1]['data'].get('sentiment_distribution'))) if ed['skeptical'] == None: ed['skeptical'] = s else: if (s > 0 and ed['skeptical'] > 0) or (s < 0 and ed['skeptical'] < 0): ed['skeptical'] = min(s, ed['skeptical']) else: ed['skeptical'] = 0 break def mdai2020_VAF_valuation(self, args): ''' Valuation function for a Valued Argumentation Framework (VAF) over UDebG ''' if not args.user_valuation: sys.exit('ERROR: no valuation function selected for the VAF over UDebG.') for n, nd in self.UDebG.nodes(data = True): if args.user_valuation == 'comment_karma': nd['valuation'] = scale_weight(max([int(c.get(args.user_valuation)) for c in nd['data']]), args) elif args.user_valuation == 'sum_scores': nd['valuation'] = scale_weight(sum([int(c.get('score')) for c in nd['data']]), args) vals = [nd['valuation'] for n_id, nd in self.UDebG.nodes(data = True)] self.max_weight = max(vals) self.min_weight = min(vals) pos_vals = [ed['skeptical'] for u1, u2, ed in self.UDebG.edges(data = True) if ed['skeptical'] > 0] neg_vals = [ed['skeptical'] for u1, u2, ed in self.UDebG.edges(data = True) if ed['skeptical'] < 0] self.max_pos_edge_weight = max(pos_vals) self.min_pos_edge_weight = min(pos_vals) self.max_neg_edge_weight = max(neg_vals) self.min_neg_edge_weight = min(neg_vals) def mdai2020_UDebG_to_xml(self, args): ''' Saves self.UDebG graph to xml file ''' xml = ET.Element('entailment-corpus') xml.append(ET.Comment(reddit_at.args2str(args))) xml.set('num_nodes', str(len(self.UDebG))) xml.set('num_edges', str(self.UDebG.number_of_edges())) al_xml = ET.SubElement(xml, 'argument-list') al_xml.set('minweight', str(self.min_weight)) al_xml.set('maxweight', str(self.max_weight)) for n_id, nd in self.UDebG.nodes(data = True): a = ET.SubElement(al_xml, 'arg') a.set('weight', str(nd['valuation'])) a.set('user', str(n_id)) a.set('id', str(nd['node_id'])) ap_xml = ET.SubElement(xml, 'argument-pairs') for u1, u2, ed in self.UDebG.edges(data = True): if ed['skeptical'] < 0 and abs(ed['skeptical']) > args.alpha: p = ET.SubElement(ap_xml, 'pair') p.set('entailment', 'ATTACKS') t = ET.SubElement(p, 't') t.set('id', str(self.UDebG.nodes[u1]['node_id'])) h = ET.SubElement(p, 'h') h.set('id', str(self.UDebG.nodes[u2]['node_id'])) ET.ElementTree(xml).write("%s.udebg.xml" % args.input_file) def mdai2020_draw_DebT(self, args): ''' Drawing Debate Tree ''' print('Drawing DebT...') gv = networkx.nx_agraph.to_agraph(self.DebT) gv.node_attr['style'] = 'filled' gv.node_attr['fixedsize'] = 'true' gv.node_attr['width'] = '0.4' gv.node_attr['height'] = '0.4' gv.node_attr['fillcolor'] = '#0000FF' gv.node_attr['fontcolor'] = '#FFFFFF' for n in gv.nodes(): n.attr['label'] = str(self.DebT.nodes[n]['chrono_id']) gv.edge_attr['color'] = '#000000' for e in gv.edges(): s = sentiment(ast.literal_eval(self.DebT.nodes[e[0]]['data'].get('sentiment_distribution'))) if s > 0: e.attr['color'] = '#00FF00' elif s < 0: e.attr['color'] = '#FF0000' gv.layout(prog = 'dot', args='-Goverlap=false -Gnodesep=0.2 -Granksep=0.2 -Grankdir=BT -GK=800 -Gstart=17 -Gmaxiter=600') gv.draw("%s.debt.png" % args.input_file, format = 'png') def mdai2020_draw_UDebG(self, args): ''' Drawing UDebG ''' if self.VAF_accepted: print('Drawing UDebG solution...') output_file_name = '%s.udebg-sol.png' % args.input_file else: print('Drawing UDebG...') output_file_name = '%s.udebg.png' % args.input_file gv = networkx.nx_agraph.to_agraph(self.UDebG) gv.node_attr['style'] = 'filled' gv.node_attr['fixedsize'] = 'true' gv.node_attr['width'] = '0.4' gv.node_attr['height'] = '0.4' for n in gv.nodes(): node_id = self.UDebG.nodes[n]['node_id'] n.attr['label'] = str(node_id) bordercolor = [0x00, 0x00, 0x00] penwidth = 1 fontcolor = '#FFFFFF' fillcolor = [0x00, 0x00, 0xFF] if self.VAF_accepted: fontcolor, fillcolor = get_weighted_color([0x00, 0x00, 0xFF], self.min_weight, self.max_weight, self.UDebG.nodes[n]['valuation']) if node_id not in self.VAF_accepted: bordercolor = fillcolor penwidth = 3 fontcolor, fillcolor = get_weighted_color([0x00, 0x00, 0x00], self.min_weight, self.max_weight, self.UDebG.nodes[n]['valuation']) n.attr['fontcolor'] = fontcolor n.attr['fillcolor'] = '#%s' % ''.join([c[2:].zfill(2) for c in map(hex, fillcolor)]) n.attr['color'] = '#%s' % ''.join([c[2:].zfill(2) for c in map(hex, bordercolor)]) n.attr['penwidth'] = penwidth gv.edge_attr['color'] = '#000000' for e in gv.edges(): if self.UDebG[e[0]][e[1]]['skeptical'] > 0: contrast, color = get_weighted_color([0x00, 0xFF, 0x00], self.min_pos_edge_weight, self.max_pos_edge_weight, self.UDebG[e[0]][e[1]]['skeptical']) e.attr['color'] = '#%s' % ''.join([c[2:].zfill(2) for c in map(hex, color)]) elif self.UDebG[e[0]][e[1]]['skeptical'] < 0: contrast, color = get_weighted_color([0xFF, 0x00, 0x00], -self.max_neg_edge_weight, -self.min_neg_edge_weight, -self.UDebG[e[0]][e[1]]['skeptical']) e.attr['color'] = '#%s' % ''.join([c[2:].zfill(2) for c in map(hex, color)]) if self.VAF_accepted: if abs(self.UDebG[e[0]][e[1]]['skeptical']) > args.alpha: e.attr['color'] = '#FF0000' else: e.attr['color'] = 'transparent' # Like do not draw edge gv.layout(prog = 'dot', args='-Goverlap=false -Gnodesep=0.2 -Granksep=0.2 -Grankdir=BT -GK=800 -Gstart=17 -Gmaxiter=600') gv.draw(output_file_name, format = 'png')