-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathserver.py
145 lines (118 loc) · 4.24 KB
/
server.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
import json
import math
import os
import socket
def main():
funcHashmap = {
'floor' : floor,
'nroot' : nroot,
'reverse' : reverse,
'validAnagram' : validAnagram,
'sort' : sort
}
paramCheckHashmap = {
'floor' : 'double',
'nroot' : '[int,int]',
'reverse' : 'string',
'validAnagram' : '[string,string]',
'sort' : 'string[]'
}
# ソケットを作成する
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
# アドレスを定義する
serverAddress = '127.0.0.1'
try:
# 前回の実行でソケットファイルが残っていた場合、そのファイルを削除する
os.unlink(serverAddress)
except FileNotFoundError:
pass
# サーバのアドレスをソケットに紐付ける
sock.bind(serverAddress)
# ソケットが接続要求を待機する
sock.listen(1)
# クライアントからの接続を受け入れる
connection, _ = sock.accept()
try:
while True:
print('--------------------')
# クライアントのソケットからのデータ(json形式)を受信する
data = connection.recv(1024)
# 受け取ったデータはバイナリ形式なので、それを文字列に変換する
dataStr = data.decode('utf-8')
# 受信したデータを表示する
print('received data : {}'.format(dataStr))
receivedData = json.loads(data)
method = receivedData['method']
params = receivedData['params']
paramTypes = receivedData['param_types']
id = receivedData['id']
# JSON形式のデータ(結果、結果の型、同じリクエストID)をクライアントに送り返す
if method in funcHashmap:
# パラメータの検証を行う
if paramCheckHashmap[method] != str(paramTypes):
sendData = 'Invalid parameter'
else:
results = funcHashmap[method](params)
resultsType = resultTypeCondition(results)
sendHashmap = {
'results' : str(results),
'result_type' : resultsType,
'id' : id
}
sendData = json.dumps(sendHashmap)
else:
sendData = 'Function not found'
connection.sendall(sendData.encode('utf-8'))
print('sent data : {}'.format(sendData))
finally:
# 最後にソケットを閉じてリソースを解放する
print('closing socket')
sock.close()
# typeを調べる
def resultTypeCondition(t):
if isinstance(t, bool):
return 'bool'
elif isinstance(t, int):
return 'int'
elif isinstance(t, float):
return 'double'
elif isinstance(t, str):
return 'str'
elif isinstance(t, list):
return 'list'
else:
return "It was a type I wasn't expecting"
# 10進数xを最も近い整数に切り捨て、その結果を整数で返す
def floor(x):
return int(math.floor(x))
# 方程式r^n=xにおける、rの値を計算する
def nroot(arr):
return math.pow(arr[1],1/arr[0])
# 入力文字列の逆である新しい文字列を返す
def reverse(s):
return s[::-1]
# 2つの入力文字列が互いにアナグラムであるかどうかを示すブール値を返す
# ここでは大文字と小文字の区別はつけずスペースは文字として扱わないものとする
def validAnagram(strArr):
s1 = strArr[0].replace(' ', '').lower()
s2 = strArr[1].replace(' ', '').lower()
hashmap1 = {}
hashmap2 = {}
if len(s1) != len(s2):
return False
for i in s1:
if i not in hashmap1:
hashmap1[i] = 1
else:
hashmap1[i] += 1
for i in s2:
if i not in hashmap2:
hashmap2[i] = 1
else:
hashmap2[i] += 1
return hashmap1 == hashmap2
# 入力である文字列の配列をソートして、ソート後の文字列rの配列を返す
def sort(strArr):
return sorted(strArr)
if __name__ == '__main__':
main()