광주인력개발원
| 미니 의료 정보 시스템 | - 3일차 - | 개발일지 [2023-07-20 학습일지]
플광
2023. 7. 30. 12:50
금일 목표 : CRUD 로직 완성, 더미 데이터 입력하기, UI-클라이언트-네트워크 로직 시작
실제 달성 : CRUD 로직 일부 완성, 더미 데이터 완료, UI-클라이언트-네트워크 로직 진행중
평가:
비효율적으로 필요없는 CRUD 로직은 작성하지 않으려고 계획중이다. (필요할 때마다 그때 그때 생성중)
DTO 를 활용하여 기존 CRUD 로직을 최대한 활용하려고 하는데, 매번 Object로 맵핑한다는 것은 굉장한 TRIAL이다.
토요일 자정까지 마감기한이 늘어났다. 로직을 깔끔하게 짜서 시간을 아끼려고 함에도,
구현해야할 분량이 여태 한 프로젝트 중 가장 많은 것 같다...
# DBConnector.py
# ================ CREATE TABLE ============================ #
def create_tables(self):
c = self.start_conn()
pstmt = None
with open(self.SQL_PATH, 'r', encoding='utf-8') as file:
pstmt = file.read()
if pstmt is None:
raise 'sql문 못불러옴'
c.executescript(pstmt)
self.commit_db()
self.end_conn()
print('created_tables')
# ===================== DOMAIN ============================ #
# ================ EMPLOYEE ============================ #
def insert_employee(self, employee: Employee):
assert isinstance(employee, Employee)
c = self.start_conn()
if employee.employee_id is None:
# insert
pstmt, values, = self.make_prepared_statement(employee, 'tb_employee')
c.execute(pstmt, values)
else:
# todo: update 로직 필요하면 작성
return
self.commit_db()
row = c.execute('select * from tb_employee order by employee_id desc limit 1').fetchone()
result = Employee(*row)
self.end_conn()
return result
def find_employee_by_id(self, employee_id):
fetched_row = self.find_obj_by_id_and_table_name(employee_id, "tb_employee", "employee")
return Employee(*fetched_row)
def find_all_employee(self):
fetched_rows = self.find_all_by_table_name("tb_employee")
result_list = list()
for row in fetched_rows:
result_list.append(Employee(*row))
return result_list
def assert_login_username_and_password(self, username, password):
c = self.start_conn()
fetched_row = c.execute(f'select * from tb_employee where login_username=? and login_password=?',
(f'{username}', f'{password}',)).fetchone()
if fetched_row is None:
return None
return Employee(*fetched_row)
# ================ DOCTOR / NURSE / ADMIN ============================ #
def insert_doctor_job(self, employee, assigned_department_id):
c = self.start_conn()
c.execute(f'''insert into tb_doctor(employee_id, assigned_department_id)
values (?, ?)''', (employee.employee_id, assigned_department_id,))
self.commit_db()
self.end_conn()
def insert_nurse_job(self, employee, assigned_ward_id):
c = self.start_conn()
c.execute(f'''insert into tb_nurse(employee_id, assigned_ward_id)
values (?, ?)''', (employee.employee_id, assigned_ward_id,))
self.commit_db()
self.end_conn()
def insert_admin_job(self, employee, assigned_part_id):
c = self.start_conn()
c.execute(f'''insert into tb_administration(employee_id, assigned_part_id)
values (?, ?)''', (employee.employee_id, assigned_part_id,))
self.commit_db()
self.end_conn()
# ================ CHAT ROOM ============================ #
def insert_chat_room(self, chat_room: ChatRoom):
assert isinstance(chat_room, ChatRoom)
c = self.start_conn()
if chat_room.chat_room_id is None:
# insert
pstmt, values, = self.make_prepared_statement(chat_room, 'tb_chat_room')
c.execute(pstmt, values)
else:
# todo: update 로직 필요하면 작성
pass
self.commit_db()
row = c.execute('select * from tb_chat_room order by chat_room_id desc limit 1').fetchone()
result = ChatRoom(*row)
self.end_conn()
return result
def find_chat_room_by_id(self, chat_room_id):
assert isinstance(chat_room_id, int)
fetched_row = self.find_obj_by_id_and_table_name(chat_room_id, "tb_chat_room", "chat_room")
# 관련 톡방 사용자(2명) list 가져오기
employee_list = self.find_employees_by_chat_room_id(chat_room_id)
# message_list 가져오기
message_list = self.find_messages_by_chat_room_id(chat_room_id)
return ChatRoom(*fetched_row, employee_list=employee_list, message_list=message_list)
def find_employees_by_chat_room_id(self, chat_room_id):
c = self.start_conn()
fetched_rows = c.execute(f"""
select * from tb_employee where employee_id in
(select employee_id from tb_employee_chat_room where chat_room_id = {chat_room_id})
""").fetchall()
result_list = list()
for row in fetched_rows:
result_list.append(Employee(*row))
self.end_conn()
return result_list
# ================ EMPLOYEE CHAT ROOM ============================ #
def insert_employee_chat_room(self, employee_id, chat_room_id):
assert isinstance(employee_id, int) and isinstance(chat_room_id, int)
c = self.start_conn()
c.execute('insert into tb_employee_chat_room(employee_id, chat_room_id) values (?, ?)',
(employee_id, chat_room_id))
self.commit_db()
self.end_conn()
# ================ MESSAGE ============================ #
def insert_message(self, message: Message):
assert isinstance(message, Message)
c = self.start_conn()
if message.message_id is None:
# insert
pstmt, values, = self.make_prepared_statement(message, 'tb_message')
c.execute(pstmt, values)
else:
# todo: update 로직 필요하면 작성
pass
self.commit_db()
row = c.execute('select * from tb_message order by message_id desc limit 1').fetchone()
result = Message(*row)
self.end_conn()
return result
def find_messages_by_message_id(self, message_id):
c = self.start_conn()
fetched_row = c.execute(f"""
select * from tb_message where message_id = {message_id})
""").fetchone()
result = Message(*fetched_row)
self.end_conn()
return result
def find_messages_by_chat_room_id(self, chat_room_id):
c = self.start_conn()
fetched_rows = c.execute(f"""
select * from tb_message where message_id in
(select employee_id from tb_employee_chat_room where chat_room_id = {chat_room_id})
""").fetchall()
result_list = list()
for row in fetched_rows:
result_list.append(Message(*row))
self.end_conn()
return result_list
def checked_message_by_chat_room_id(self, chat_room_id):
c = self.start_conn()
fetched_rows = c.execute(f"""
update tb_message set "is_confirmed"=1 where message_id in
(select employee_id from tb_employee_chat_room where chat_room_id = {chat_room_id})
""").fetchall()
result_list = list()
for row in fetched_rows:
result_list.append(Message(*row))
self.end_conn()
return result_list아직 로그인 기능만 구현한 서버
class EMRServer:
def __init__(self, db_connector: DBConnector):
self.db_conn = db_connector
self.common = Common()
self.server_socket = None
self.sockets_list = list()
self.clients = dict()
self.thread_for_run = None
self.run_signal = True
self.decoder = ObjDecoder()
self.encoder = ObjEncoder()
self.dto_maker = DTOMaker()
def start(self):
if self.thread_for_run is not None: # 실행중이면 종료 시키기
return
self.server_socket = socket(AF_INET, SOCK_STREAM) # AF_INET(ipv4를 의미)
self.server_socket.bind((self.common.HOST, self.common.PORT)) # 바인딩
self.server_socket.listen() # 리슨 시작
self.sockets_list.clear() # 소켓리스트 클리어
self.sockets_list.append(self.server_socket)
self.run_signal = True
self.thread_for_run = Thread(target=self.run)
self.thread_for_run.start()
print('SERVER STARTED')
def run(self):
while True:
if self.run_signal is False:
break
try:
read_sockets, _, exception_sockets = select.select(self.sockets_list, [], self.sockets_list, 0.1)
except Exception:
continue
for notified_socket in read_sockets:
if notified_socket == self.server_socket:
client_socket, client_address = self.server_socket.accept()
user = self.receive_message(client_socket)
if user is False:
continue
self.sockets_list.append(client_socket)
self.clients[client_socket] = user
else:
message = self.receive_message(notified_socket)
if message is False:
self.sockets_list.remove(notified_socket)
del self.clients[notified_socket]
continue
for notified_socket in exception_sockets:
self.sockets_list.remove(notified_socket)
del self.clients[notified_socket]
def send_message(self, client_socket:socket, header, data):
assert isinstance(header, str) or isinstance(header, bytes)
assert isinstance(data, str) or isinstance(data, bytes)
if isinstance(header, bytes):
header_to_sending = header.decode(self.common.FORMAT)
else:
header_to_sending = header
if isinstance(data, bytes):
data_to_sending = data.decode(self.common.FORMAT)
else:
data_to_sending = data
message_to_send = f"{self.common.START_OF_TEXT}".join([header_to_sending, data_to_sending]) \
.encode(self.common.FORMAT)
if len(message_to_send) >= self.common.BUFFER:
raise "부족한 버퍼 용량"
print(f"SERVER SENDED: (HEADER: {header} | DATA: {data})")
client_socket.send(message_to_send)
def header_data_distributor(self, recv_encoded_message):
decoded_message = recv_encoded_message.decode(self.common.FORMAT).strip()
header, data_decoded_str, = decoded_message.split(self.common.START_OF_TEXT)
data = self.decoder.binary_to_obj(data_decoded_str)
return header, data
def receive_message(self, client_socket: socket):
while True:
try:
recv_encoded_message = client_socket.recv(self.common.BUFFER)
request_header, request_data = self.header_data_distributor(recv_encoded_message)
print(f"SERVER RECEIVED: ({request_header},{request_data})")
except Exception:
return False
if request_header == self.common.LOGIN_ACCESS_REQ:
username = request_data['login_username']
password = request_data['password']
login_employee = self.db_conn.assert_login_username_and_password(username, password)
if login_employee is None:
self.send_message(client_socket, self.common.LOGIN_ACCESS_RES,self.common.FALSE)
else:
# 로그인 성공 로직
# 회원 정보 보내기
data_str = self.encoder.toJSON_an_object(login_employee)
self.send_message(client_socket, self.common.LOGIN_ACCESS_RES, data_str)오늘 작성한 SQL, 일부 더미데이터를 추가했다 내일 몇가지 관계를 추가해야해서 머리가 아프다...;
DROP TABLE IF EXISTS tb_ktas;
CREATE TABLE "tb_ktas" (
"ktas_id" INTEGER,
"first_category_name" TEXT NOT NULL,
"first_category_code" TEXT NOT NULL,
"second_category_name" TEXT NOT NULL,
"second_category_code" TEXT NOT NULL,
"third_category_name" TEXT NOT NULL,
"third_category_code" TEXT NOT NULL,
"fourth_category_name" TEXT NOT NULL,
"fourth_category_code" TEXT NOT NULL,
"final_grade" TEXT NOT NULL,
"ktas_code" TEXT NOT NULL,
PRIMARY KEY("ktas_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS tb_employee;
CREATE TABLE "tb_employee" (
"employee_id" INTEGER,
"name" TEXT NOT NULL,
"type_job" INTEGER NOT NULL,
"login_username" TEXT NOT NULL UNIQUE,
"login_password" TEXT NOT NULL,
"mobile_phone_num_1" TEXT,
"mobile_phone_num_2" TEXT,
PRIMARY KEY("employee_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS tb_department;
CREATE TABLE "tb_department" (
"department_id" INTEGER,
"name" TEXT NOT NULL,
"job_category" INTEGER NOT NULL,
PRIMARY KEY("department_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS tb_doctor;
CREATE TABLE "tb_doctor" (
"doctor_id" INTEGER,
"employee_id" INTEGER NOT NULL,
"assigned_department_id" INTEGER NOT NULL,
PRIMARY KEY("doctor_id" AUTOINCREMENT)
FOREIGN KEY("assigned_department_id") REFERENCES "tb_department"("department_id")
);
DROP TABLE IF EXISTS tb_nurse;
CREATE TABLE "tb_nurse" (
"nurse_id" INTEGER,
"employee_id" INTEGER NOT NULL,
"assigned_ward_id" INTEGER NOT NULL,
PRIMARY KEY("nurse_id" AUTOINCREMENT),
FOREIGN KEY("employee_id") REFERENCES "tb_employee"("employee_id")
FOREIGN KEY("assigned_ward_id") REFERENCES "tb_department"("department_id")
);
DROP TABLE IF EXISTS tb_administration;
CREATE TABLE "tb_administration" (
"administration_id" INTEGER,
"employee_id" INTEGER NOT NULL,
"assigned_part_id" INTEGER NOT NULL,
PRIMARY KEY("administration_id" AUTOINCREMENT),
FOREIGN KEY("employee_id") REFERENCES "tb_employee"("employee_id"),
FOREIGN KEY("assigned_part_id") REFERENCES "tb_department"("department_id")
);
DROP TABLE IF EXISTS tb_chat_room;
CREATE TABLE "tb_chat_room" (
"chat_room_id" INTEGER,
"created_time" TEXT NOT NULL,
PRIMARY KEY("chat_room_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS tb_employee_chat_room;
CREATE TABLE "tb_employee_chat_room" (
"employee_chat_room_id" INTEGER,
"chat_room_id" INTEGER NOT NULL,
"employee_id" INTEGER NOT NULL,
FOREIGN KEY("chat_room_id") REFERENCES "tb_chat_room"("chat_room_id"),
FOREIGN KEY("employee_id") REFERENCES "tb_employee"("employee_id"),
PRIMARY KEY("employee_chat_room_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS tb_patient;
CREATE TABLE "tb_patient" (
"patient_id" INTEGER,
"birth_date" TEXT,
"name" TEXT NOT NULL,
"sex" TEXT NOT NULL,
"ssn" TEXT UNIQUE,
"address" TEXT,
"type_insurance" TEXT,
"allocated_bed_location_id" INTEGER,
"register_number" INTEGER UNIQUE,
FOREIGN KEY("allocated_bed_location_id") REFERENCES "tb_bed_of_ward"("bed_id"),
PRIMARY KEY("patient_id" AUTOINCREMENT)
);
DROP TABLE IF EXISTS tb_bed_of_ward;
CREATE TABLE "tb_bed_of_ward" (
"bed_id" INTEGER,
"assigned_ward_id" INTEGER NOT NULL,
"additional_info" TEXT,
"viewing_bed_name" TEXT NOT NULL,
PRIMARY KEY("bed_id" AUTOINCREMENT),
FOREIGN KEY("assigned_ward_id") REFERENCES "tb_department"
);
DROP TABLE IF EXISTS tb_emergency_nurse_record;
CREATE TABLE "tb_emergency_nurse_record" (
"enr_id" INTEGER,
"register_id" INTEGER,
"onset_time" TEXT,
"ktas_id" INTEGER,
"cheif_complain" TEXT,
"description" TEXT,
"recorder_nurse_id" INTEGER,
"saved_time" TEXT,
"responser" TEXT,
PRIMARY KEY("enr_id" AUTOINCREMENT),
FOREIGN KEY("recorder_nurse_id") REFERENCES "tb_nurse"("nurse_id")
FOREIGN KEY("register_id") REFERENCES "tb_patient"("register_number")
);
DROP TABLE IF EXISTS tb_message;
CREATE TABLE "tb_message" (
"message_id" INTEGER,
"sender_employee_id" INTEGER NOT NULL,
"chat_room_id" INTEGER NOT NULL,
"contents" TEXT NOT NULL,
"is_confirmed" INTEGER NOT NULL,
FOREIGN KEY("chat_room_id") REFERENCES "tb_chat_room"("chat_room_id"),
PRIMARY KEY("message_id" AUTOINCREMENT),
FOREIGN KEY("sender_employee_id") REFERENCES "tb_employee"("employee_id")
);
INSERT INTO tb_department("name", "job_category") VALUES
("응급의학과", 1),("신경외과", 1), ("흉부외과", 1), ("외과", 1), ("내과", 1), ("소아청소년과", 1),
("7A",2),("9B",2),("ER",2),("MICU",2),("101W",2),("EICU",2),
("원무과",3),("보험심사팀",3);
INSERT INTO tb_bed_of_ward("assigned_ward_id", "additional_info", "viewing_bed_name") VALUES
(9, "소생실", "CR1"),
(9, "소생실", "CR2"),
(9, "중증구역", "IC1"),
(9, "중증구역", "IC2"),
(9, "중증구역", "IC3"),
(9, "중증구역", "IC4"),
(9, "중증구역", "IC5"),
(9, "중증구역", "IC6"),
(9, "중증구역", "IC7"),
(9, "중증구역", "IC8"),
(9, "중증구역", "IC9"),
(9, "중증구역", "IC10"),
(9, "중증구역", "IC11"),
(9, "중증구역", "IC12"),
(9, "경증구역", "TR");
INSERT INTO tb_patient("birth_date", "name","sex", "ssn", "address", "type_insurance", "allocated_bed_location_id", "register_number") VALUES
('1958-06-08', '김철수', 'M', '580608-1354644', '경기도 수원시 언저리 어딘가', '건강보험', 1, 486124);
INSERT INTO tb_emergency_nurse_record(
"onset_time",
"register_id",
"ktas_id",
"cheif_complain",
"description",
"recorder_nurse_id",
"saved_time",
"responser") values
('1958-06-08',486124, 'AICAB', '가슴이 답답해요',
"내원전 2시간 전, 속 울렁거림, 가슴 답답함 느껴졌으나 시간 지나니 괜찮아짐, 식사 후 집으로 귀가하던 중 가슴 부여잡은 채 쓰러진 것 행인이 목격, 119 신고로 이송됨 이송 중에 의식 깨어 현재는 alert하나 흉통(NRS>7) 계속 호소중임",
1, "2023-07-20 13:11:22", "환자 본인");더미 데이터 추가 로직
def insert_dummy_employee_data(conn: DBConnector, size):
# 우선 내 거 하나
conn.insert_employee(Employee(None, "박광현", 2, "qwer11", "1234", "010-1010-0102", "010-0215-1335"))
maker = FakeDataMaker()
for i in range(size):
# 가짜 생성
name = maker.get_random_fake_name()
type_job = maker.get_random_type_job()
login_username = maker.get_random_login_id()
login_pw = "1234"
phone_1 = maker.get_random_phone_num()
phone_2 = maker.get_random_phone_num()
# 객체화 및 입력
dummy_employee = Employee(None, name, type_job, login_username, login_pw, phone_1, phone_2)
dummy_employee = conn.insert_employee(dummy_employee)
if dummy_employee.type_job == 1:
conn.insert_doctor_job(dummy_employee, random.randint(1, 6))
elif dummy_employee.type_job == 2:
conn.insert_nurse_job(dummy_employee, random.randint(7, 12))
elif dummy_employee.type_job == 3:
conn.insert_admin_job(dummy_employee, random.randint(13, 14))
def insert_dummy_chat_room(conn: DBConnector, size):
maker = FakeDataMaker()
for i in range(size):
# 가짜 생성
# 객체화 및 입력
random_date = maker.get_random_date_time_str()
dummy_chat_room = ChatRoom(None, random_date)
conn.insert_chat_room(dummy_chat_room)
def insert_dummy_employee_chat_room_data(conn: DBConnector, size, employee_range, chat_range):
maker = FakeDataMaker()
min_id_employee, max_id_employee = employee_range
min_id_chat, max_id_chat = chat_range
for i in range(size):
# 가짜 생성
# 객체화 및 입력
random_employee_id_1, random_employee_id_2 = random.sample(range(min_id_employee, max_id_employee), 2)
random_chat_room_id = random.randint(min_id_chat, max_id_chat)
conn.insert_employee_chat_room(random_employee_id_1, random_chat_room_id)
conn.insert_employee_chat_room(random_employee_id_2, random_chat_room_id)
for _ in range(random.randint(1, 10)):
# 가짜 생성
# 객체화 및 입력
dummy_message_1 = Message(None, random_employee_id_1, random_chat_room_id, maker.get_random_lorem(), False)
dummy_message_2 = Message(None, random_employee_id_2, random_chat_room_id, maker.get_random_lorem(), False)
conn.insert_message(dummy_message_1)
conn.insert_message(dummy_message_2)
def insert_dummy_message_data(conn: DBConnector, size, employee_range, chat_range):
maker = FakeDataMaker()
min_id_employee, max_id_employee = employee_range
min_id_chat, max_id_chat = chat_range
for i in range(size):
# 가짜 생성
# 객체화 및 입력
random_employee_id = random.randint(min_id_employee, max_id_employee)
random_chat_room_id = random.randint(min_id_chat, max_id_chat)
dummy_message = Message(None, random_employee_id, random_chat_room_id, maker.get_random_lorem(), False)
conn.insert_message(dummy_message)
내일은 UI 구현, 실시간 채팅을 구현할 예정이다.
감사합니다.