Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Slack #6

Open
wants to merge 7 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -1 +1,2 @@
data/idm.csv
members_ID.csv
Binary file added __pycache__/hook.cpython-39.pyc
Binary file not shown.
Binary file added __pycache__/post_alert_to_slack.cpython-39.pyc
Binary file not shown.
Binary file added lib/__pycache__/lcd.cpython-39.pyc
Binary file not shown.
18 changes: 14 additions & 4 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,21 @@
import pigpio
from time import sleep
import time

import datetime
import pandas as pd
import datetime

import lib.lcd as lcd
from lib.lcd import lcd_byte, lcd_string, lcd_init
from hook import open_door_hook, close_door_hook


from post_alert_to_slack import check_open


SCHEDULE_LINK = "https://cal.camph.net/public/schedule.json"


class Servo:
#set servo PIN
servo_pin = 18
Expand Down Expand Up @@ -90,14 +97,15 @@ def registering_card(self,idm):
self.df.to_csv('/home/pi/HIKIJOH/data/idm.csv')



def main():
global LCD_BACKLIGHT

door_isopen = True
want_close = False
card_istouch = False

check_open(door_isopen=door_isopen)

card = NFC_CARD()
servo = Servo()
nfc_data = NFC_Data()
Expand Down Expand Up @@ -162,7 +170,7 @@ def main():
else:
lcd_string(" THIS CARD IS ", lcd.LCD_LINE_1)
lcd_string(" NOT MASTER CARD ", lcd.LCD_LINE_2)

if time.time() - card_touched_time > 5:
is_timeout = True

Expand Down Expand Up @@ -190,8 +198,10 @@ def main():
#LCDの焼け付きを防止するために消す
want_close = False
LCD_BACKLIGHT = 0x00 #バックライトオフ
lcd_byte(0x01, lcd.LCD_CMD) #表示内容クリア
lcd_byte(0x01, lcd.LCD_CMD) #表示内容クリア

# 開館の20分前から3時間後までの間にドアが開いていない場合はSlackに通知
check_open(door_isopen=door_isopen)


if __name__ == '__main__':
Expand Down
191 changes: 191 additions & 0 deletions post_alert_to_slack.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,191 @@
import textwrap
from datetime import datetime, timedelta, tzinfo
import os
from time import sleep
from typing import Dict, List, Optional

import csv
import dateutil.parser
import pytz
import requests
from requests.auth import HTTPBasicAuth


WEEKDAY_NAMES = ['月', '火', '水', '木', '金', '土', '日']
TIME_ZONE = "Asia/Tokyo"

# カレンダー
CAL_USER = os.environ["CAL_USER"]
CAL_PASSWORD = os.environ["CAL_PASSWORD"]
CAL_LINK = "https://cal.camph.net/private/schedule.json"

# Slack_API
SLACK_TOKEN = os.environ["SLACK_TOKEN"]
SLACK_CHANNEL = os.environ["SLACK_CHANNEL"]
SLACK_LINK = "https://slack.com/api/chat.postMessage"

class Event:
start: datetime
end: datetime
title: str
opener: str
url: Optional[str]

def __init__(self, *, start: datetime, end: datetime, title: str, opener: str,
url: Optional[str]) -> None:
self.start = start
self.end = end
self.title = title
self.opener = opener
self.url = url

@classmethod
def from_json(cls, data: Dict[str, Optional[str]]) -> "Event":
# print(type(data["title"]))
if data["start"] is not None:
start = dateutil.parser.parse(data["start"])
else:
raise KeyError
if data["end"] is not None:
end = dateutil.parser.parse(data["end"])
else:
raise KeyError
if data["title"] != "Closed":
title = "open"
opener = list(data["title"].split())
elif data["title"] == "Closed":
title = "closed"
opener = []
else:
raise KeyError
return cls(start=start, end=end, title=title, opener=opener, url=data["url"])

def get_start(self, tz: tzinfo) -> str:
start = self.start.astimezone(tz).time().strftime("%H:%M")
return start

def get_end(self, tz: tzinfo) -> str:
end = self.end.astimezone(tz).time().strftime("%H:%M")
return end

def get_day_and_time(self, tz: tzinfo) -> str:
date = self.start.astimezone(tz).date().strftime("%m/%d")
day = get_japanese_weekday(self.start.astimezone(tz).weekday())
return f"{date} ({day}) {self.get_start(tz)}〜{self.get_end(tz)}"

def get_day_and_time_with_title(self, tz: tzinfo) -> str:
return f"{self.title} {self.get_day_and_time(tz)}"


def divide_events_by_title(events: List[Event]) -> Dict[str, List[Event]]:
events_by_title: Dict[str, List[Event]] = {}
for title in ["open", "make", "online open"]:
events_by_title[title] = [
e for e in events if e.title.lower() == title]
events = [e for e in events if e.title.lower() != title]
events_by_title["other"] = [e for e in events if e.title.strip() != ""]
return events_by_title


def get_japanese_weekday(day: int) -> str:
return WEEKDAY_NAMES[day]


def download_events(url: str) -> Optional[List[Event]]:
# BASIC認証
auth = HTTPBasicAuth(CAL_USER, CAL_PASSWORD)
response = requests.get(url, auth=auth)
if response.status_code != requests.codes.ok:
return None
return [Event.from_json(e) for e in response.json()]


def post_alert_to_slack(message: str) -> None:
url = SLACK_LINK
headers = {"Authorization": "Bearer " + SLACK_TOKEN}
data = {
"channel": SLACK_CHANNEL,
"text": message,
}

r = requests.post(url=url, headers=headers, data=data)
print('return', r.json())


def check_open(door_isopen: bool):
tz = pytz.timezone(TIME_ZONE)
now = datetime.now(tz=tz)
# if now is None:
# now = datetime.now(tz=tz)
# elif now.tzinfo is None:
# now = now.replace(tzinfo=tz)

# イベントの取得
events = download_events(CAL_LINK)
events = list(filter(
lambda e: e.start.astimezone(tz).date()
== now.astimezone(tz).date(), events))
events_by_title = divide_events_by_title(events)

# 時間の取得
open_events = events_by_title["open"]
if len(open_events) == 1:
open_event = open_events[0]
open_start = open_event.get_start(tz)
# CAMPHOR- Make あり
make_events = events_by_title["make"]
if len(make_events) == 1:
make_event = make_events[0]
make_start = make_event.get_start(tz)
elif len(make_events) > 1:
raise ValueError("The maximum number of Make events per day"
" is one, but found several.")
elif len(open_events) > 1:
raise ValueError("The maximum number of Open events per day is"
" one, but found several.")
elif len(open_events) == 0:
open_start = None
make_start = None

# テスト
open_start = "15:00"
opener = "@Tomo"

if open_start is None:
pass
else:
# 開館の20分前から3時間後までの間にドアが開いていない場合はSlackに通知
# try:
# opener = ''.join(events_by_title["open"][0].opener)
# except:
# raise ValueError("Member is not found.")

# user_IDの取得
# with open('members_ID.csv') as f:
# reader = csv.reader(f)
# members = [row for row in reader]
#
# for user, id_ in members:
# if user == opener:
# id = id_

# message通知の時間設定
alert_interval_minutes = 10 * 60
before_alert_minutes = 10
before_delta = timedelta(minutes=before_alert_minutes)
after_alert_hour = 3
after_delta = timedelta(hours=after_alert_hour)

# テスト
before_delta = timedelta(hours=24)
after_delta = timedelta(hours=24)

open_start = datetime.strptime(open_start, '%H:%M')
dt_open_start = datetime.now().replace(hour=open_start.hour, minute=open_start.minute, second=0, microsecond=0).astimezone(tz)

# messageの通知
message = "@channel 開館されていません"
# テスト
# message = "<@channel>" + " 開館されていません"
# if (dt_open_start - before_delta <= now <= dt_open_start + after_delta) and (not door_isopen):
post_alert_to_slack(message=message)