分類
程序

在 VS Code 中調試 Django Q 任務

在 .vscode 文件夾中創建一個 lanch.json 文件, 內容如下:

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Django Q",
            "type": "python",
            "request": "launch",
            "program": "${workspaceFolder}/manage.py",
            "args": [
                "qcluster"
            ],
            "django": true,
            "env": {
                "DJANGO_SETTINGS_MODULE": "YOUR_PROJECT_NAME.settings"
            },
            "gevent": true
        },
        // Your other configurations go here
    ]
}

只需修改兩個地方:program 裏的 ${workspaceFolder} 是變量,如果你項目的 manage.py 在 VS Code 項目根目錄下,那這行就不用修改。 DJANGO_SETTINGS_MODULE 這裏需要把 YOUR_PROJECT_NAME 替換成你項目的名字。然後按 Debug 按鈕就可以調試 Django Q task 了。

對於這個問題,ChatGPT 4 前前後後給出了五次錯誤配置,不得不記錄一下。

分類
网站

使用 Django Q 方便地執行耗時任務

Django Q 是一個使用 Python 多進程製作的原生 Django 任务队列、调度器和 worker 应用。它具有很多優異特性但我只是粗淺使用了如下幾點:

  • 多進程 worker 池
  • 異步任務
  • 定時任務、cron 和重複的任務
  • 把失敗和成功結果保存到數據庫或緩存
  • 自動集成到 Django Admin,就可以在後臺添加和管理任務
  • 支持 Redis, Disque, IronMQ, SQS, MongoDB 或 ORM 這麼多種隊列代理方式,最方便的當然是 ORM
  • 注意:Django Q 的任務間隔粒度是 30 秒,如果你的任務頻率或精準度要求高於 30 秒,你需要嘗試修改源碼或使用其他任務隊列

安裝 Django Q

pip install django-q
#如果要使用 cron 規則,則也要安裝
pip install croniter
#在項目的 settings.py 文件 INSTALLED_APPS 裏加入 django_q
INSTALLED_APPS = (
    # other apps
    'django_q',
)
#設置代理方式,我選擇 ORM,
#只需把下面字段也加入 settings.py

#更多設置請參見 https://django-q.readthedocs.io/en/latest/configure.html
Q_CLUSTER = {
    'name': 'djangtasks',
    'workers': 2,
    'timeout': 180,
    'retry': 200,
    'queue_limit': 50,
    'bulk': 10,
    'orm': 'default'
}

#執行數據庫遷移來創建數據庫表
python manage.py migrate
#運行 Django Q 來處理任務隊列
python manage.py qcluster

我是使用 screen 來在後臺運行 qcluster,很方便的。

使用 Django Q 在 Django 後臺執行耗時任務

# file: views.py
import datetime
from django.http import HttpResponse
from django.utils import timezone
from django_q.tasks import async_task, schedule
from django_q.models import Schedule

def a_longtime_task(arg):
    time.sleep(30)
    return arg

def scheduled_task(arg):
    time.sleep(30)
    return arg

def a_longtime_task_request(request):
    #立即在後臺執行
    async_task(a_longtime_task,'args for the function')
    return HttpResponse('The longtime task has been started.')

def another_longtime_task_request(request):
    #三分鐘後執行
    schedule('YOURAPP.views.scheduled_task',
            'args for the function',
            schedule_type=Schedule.ONCE,
            next_run=timezone.now() + datetime.timedelta(minutes=3))
    return HttpResponse('The task has been scheduled.')

在 Django Admin 佈置定時任務

這個就比較直接,比如要每天 10 點都執行上面例子中的 scheduled_task,就點擊 Admin 頁面 Scheduled tasks 旁邊的「新增」。

Func(填入完整的函數路徑):YOURAPP.views.scheduled_task
Args:'args for the function'
Schedule Type:Daily
Next Run(設置要運行的時間如):2022-08-27 10:00

最後點擊「儲存」就可以了。

同樣的,如果選擇 Cron 類型,就需要在 Cron 輸入框填入 Cron 計劃。比如在 9 點到 23 點期間,每 5,15,25,35,45,55 各執行一次:
5,15,25,35,45,55 9-23 * * *
更多 Cron 用法和組合可以參考 crontab.guru 這個網站。

等待執行的任務、失敗的任務和成功的任務都可以方便的在Admin頁面查看和操作,非常方便。

管理隊列任務與已完成任務 Schedule Task

# file: tasks.py
from django_q.models import Schedule, Task
from django.db.models import Q
#your model to be checked
from app1.models import Race
from django.utils import timezone

import datetime
import operator
from functools import reduce

# Task to delete successful old tasks
def delete_old_tasks():
    len_task_20 = 0
    len_task_gen = 0
    now = datetime.datetime.now()
    days_passed_2 = timezone.utc.localize(now - datetime.timedelta(days=2))
    days_passed_7 = timezone.utc.localize(now - datetime.timedelta(days=7))
    task_q_list = []
    task_q_list.append(Q(group__startswith='20'))
    task_q_list.append(Q(started__lt=days_passed_7))
    task_q_list.append(Q(success__exact=True))
    task_20_queryset = Task.objects.filter(reduce(operator.and_, task_q_list))
    len_task_20 = len(task_20_queryset)
    for task in task_20_queryset:
        task.delete()

    task_q_list = []
    task_q_list.append(Q(group__exact='generate_0_task'))
    task_q_list.append(Q(started__lt=days_passed_2))
    task_q_list.append(Q(success__exact=True))
    task_gen_queryset = Task.objects.filter(reduce(operator.and_, task_q_list))
    len_task_gen = len(task_gen_queryset)
    for task in task_gen_queryset:
        task.delete()

    return f'{len_task_20} checked tasks and {len_task_gen} checking tasks have been removed.'

#generate 0 task every 5 minuts
def generate_0_task():
    res=[]
    now = datetime.datetime.now()
    dto_now = timezone.utc.localize(now)
    start_datetime_6 = now + datetime.timedelta(minutes=6)
    dto_plus6 = timezone.utc.localize(start_datetime_6)
    current_races = Race.objects.filter(post_time_live__lte=dto_plus6,
                                          post_time_live__gte=dto_now)
    if len(current_races) == 0:
        res.append('no current races')
        return res
    for race in current_races:
        if race.race_conditions.find('SIMULCAST')>-1:
            res.append(str(race)+': task passed SIMULCAST')
            continue
        tasks = Schedule.objects.filter(name=str(race))
        if len(tasks)==0:
            schedule('app1.views.zero_mtp_task',
                    race.id, race.track_id, race.race_number, race.race_date, race.post_time_live.isoformat(),
                    schedule_type=Schedule.ONCE,
                    next_run=race.post_time_live,
                    name=str(race))
            res.append(str(race)+': task added'+' post time:'+str(race.post_time_live))
        else:
            res.append(str(race)+': task already added'+' post time:'+str(race.post_time_live))
    return res

本文更新於 2023/01/27。

分類
网站

Django with PostgreSQL

參考 PostgreSQL:Linux downloads (Red Hat family) 來安裝PostgreSQL,比如 Fedora 35 可以這樣:

sudo dnf install -y https://download.postgresql.org/pub/repos/yum/reporpms/F-35-x86_64/pgdg-fedora-repo-latest.noarch.rpm
sudo dnf install -y postgresql13-server libpq-devel
sudo /usr/pgsql-13/bin/postgresql-13-setup initdb
sudo systemctl enable postgresql-13
sudo systemctl start postgresql-13

安裝完成後來初始化一個數據庫和用戶。

#使用 root 登錄 PostgreSQL
sudo -u postgres psql
#創建數據庫
CREATE DATABASE project;
#創建新用戶
CREATE USER user42 WITH PASSWORD 'password42';
#下面這三行優化是 Django 推薦的
ALTER ROLE user42 SET client_encoding TO 'utf8';
ALTER ROLE user42 SET default_transaction_isolation TO 'read committed';
ALTER ROLE user42 SET timezone TO 'Asia/Taipei';
#把數據庫授權給新用戶
GRANT ALL PRIVILEGES ON DATABASE project TO user42;
#退出數據庫
\q

要在 Django 中使用 PostgreSQL 數據庫,需要還安裝 psycopg2:

pip install Django psycopg2
#如果像我一樣在 Fedora 安裝失敗了
#可以嘗試安裝編譯好的版本
pip install Django psycopg2-binary

之後在項目的設置中把默認的 SQLite 數據庫資料改成 PostgreSQL 就可以了。更像詳細的步驟與解釋,可以參考:How To Use PostgreSQL with your Django Application on Ubuntu 20.04

刪除 Django 數據庫中的表格並重建

這似乎不是正確的回滾數據庫的操作辦法,但是也可以一試。

#從 migrations 目錄找到要刪除的數據庫變更文件,刪除掉
#進入數據庫操作程序
python manage.py dbshell
#查看並找到要刪除的表名
SELECT * FROM pg_catalog.pg_tables;
#如果是MariaDB,這樣查看表名
show tables;
#比如表名爲 task_day,則這樣刪除
DROP TABLE task_day;
#退出數據庫操作程序
exit;
#刪掉數據庫裏的 migrations
python manage.py migrate --prune task

#現在可以重新建立表格了
python manage.py makemigrations
python manage.py migrate

本文更新於 2022/10/24。

分類
Linux 程序

Django簡易搭建上傳文件

這裡使用Django+Gunicorn+Nginx的方式簡單運行一個小型webserver,實現一個簡單的上傳文件到服務器的功能(並不生成下載鏈接)。

啟動虛擬環境,安裝django和gunicorn:

pip install Django==2.0
pip install gunicorn
#進入要放置代碼的目錄並新建項目
django-admin startproject mysite
##或者在當前目錄建立項目
#django-admin startproject mysite .
cd mysite
#新建app
python manage.py startapp polls

先建立一個表格:

#polls/forms.py
from django import forms

class UploadFileForm(forms.Form):
    title = forms.CharField(label='密碼',max_length=20,widget=forms.PasswordInput)
    file = forms.FileField(label='文件',)

修改view:

#polls/views.py
import os
import subprocess
from django.core.files.storage import FileSystemStorage
from django.conf import settings
from django.shortcuts import render
from django.http import HttpResponse
from .forms import UploadFileForm

#handle file example with file
def handle_uploaded_file(f):
    with open('/file/should/be/saved/here/target.odt', 'wb+') as destination:
        for chunk in f.chunks():
            destination.write(chunk)

#another handle file example with filename
def handle_uploaded_file2(filename):
    msg=''
    try:
        targetZipFilePath = os.path.join(settings.BASE_DIR, filename)
        cmd1=subprocess.check_call(["unzip", "-o", targetZipFilePath, "-d", "/home/fred/workspace/"])
        if cmd1==0 :
            cmd2=subprocess.check_call(["cp", "-Rf", "/home/fred/workspace/dist", "/home/fred/"])
            if cmd2==0 :
                msg="deployed successfully"
            else:
                msg="error 2"
        else:
            msg="error 1"
    except:
        msg = 'error 0'
    return msg

def upload_file(request):
    if request.method == 'POST':
        form = UploadFileForm(request.POST, request.FILES)
        if form.is_valid() and request.POST['title']=='Secret':
#            handle_uploaded_file(request.FILES['file'])
#            return HttpResponse("上傳成功")
            myfile = request.FILES['file']
            fs = FileSystemStorage()
            filename = fs.save(myfile.name, myfile)
            uploaded_file_url = fs.url(filename)
            print(uploaded_file_url)
            res = handle_uploaded_file2(uploaded_file_url)
            return HttpResponse(res)
    else:
        form = UploadFileForm()
    return render(request, 'upload.html', {'form': form})

新建一個表格的模板:

#polls/templates/upload.html
<form enctype="multipart/form-data" action="/polls/upload/" method="post">
    {% csrf_token %}
    {{ form }}
    <input type="submit" value="上傳" />
</form>

新建一個url路由表:

#polls/urls.py
from django.urls import path

from . import views

urlpatterns = [
    path('upload/', views.upload_file, name='upload_file'),
]

修改項目路由:

#mysite/urls.py
from django.contrib import admin
from django.urls import include,path

urlpatterns = [
    path('polls/', include('polls.urls')),
    path('admin/', admin.site.urls),
]

修改項目設置:

#mysite/settings.py
INSTALLED_APPS = [    
    'polls.apps.PollsConfig',
#    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
#    'django.contrib.messages',
#    'django.contrib.staticfiles',
]
LANGUAGE_CODE = 'zh-Hant'
TIME_ZONE = 'Asia/Taipei'

然後在項目目錄(最上層)運行gunicorn就可以訪問了:

gunicorn mysite.wsgi --bind 127.0.0.1:3040

nginx中增加如下server即可在外網訪問了(鏈接應該是http://YourPublicIP:8081/polls/upload/):

server {
        listen 8081;
        server_name 127.0.0.1;
        charset utf-8;
        keepalive_timeout 60s;
        #access_log logs/django2a.access.log combined if=$loggable;
        
        location / {
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_redirect off;
    
            if (!-f $request_filename) {
                proxy_pass http://127.0.0.1:3040;
                break;
            }
    }
}

CentOS6+Django2+MySql

yum install MySQL-python
pip install mysqlclient
#如果import MySQLdb提示無libmysqlclient.so.18
#則建立軟連接如下
ln -s /usr/local/mysql/lib/libmysqlclient.so.18 /usr/lib64/libmysqlclient.so.18
分類
方法

通過 ICS 文件給日曆增加農曆

很長一段時間以來,我一直受沒有農曆可用的困擾。安卓原生的日曆不支持農曆,谷歌日曆也是到了安卓 8 才開始有農曆,等我用到安卓 8 的時候我已經開始去谷歌化了。轉用開源日曆應用(Etar)後也沒有發現很好的辦法來解決農曆的問題,主要是曆法這東西真的很複雜。終於在最近找到了一個方案,通過 ICS 文件把計算好的農曆導入日曆就可以了。我這裏提供了從 2017 年 01 月 01 日到 2049 年 12 月 31 日的農曆和節氣與公曆對應關係的 ICS 文件,只要選擇需要的年份導入日曆即可。 這些 ICS 文件是基於 infinet 開發的 lunar-calendar 項目生成的,其原始日期數據來自於香港天文台的公曆與農曆日期對照表。我在使用的時候將節氣和生肖做了繁體中文的轉換,並且刪減了一些不展示在日曆中的文本以減小文件大小。我修改後的腳本也可以在上面的分享文件夾中找到。

順便說下我這邊具體是怎麼導入的:我一直在用 Disroot 提供的 Nextcloud 實例,其中有日曆的功能,而我之前就已經在安卓上通過 DAVx⁵ 來同步日曆和聯繫人了。所以我是在電腦上通過瀏覽器來操作的,理論上在手機上也是類似的。日曆網頁的左側點「新日曆」選擇「新日曆」輸入名稱就創建好了新日曆了,如「農曆」。然後點擊左下角的「設定及匯入」-「匯入日曆」,選擇 ICS 文件然後選擇要匯入的日曆即「農曆」。最後耐心等待一番,因爲我通過網絡請求看到農曆竟然是一條一天這樣傳的,我的文件都是三年三年的,所以要耐心等它傳完,傳完後點擊一下「農曆」左側的開關就可以看到農曆了。手機上同步前先從菜單中「刷新日曆列表」就會出現「農曆」,再來同步就可以在日曆裏看到農曆和節氣了。

當然你也可以通過 Calendar Import-Export 將 ICS 文件導入本地日曆。

本文更新於 2023/03/06。

分類
网站

使用 UpUp 讓網站離線可用

UpUp 是個很小(2.5 KB)的 Javascript 庫,但卻可以方便的實現網站的離線訪問。我之前有一個小項目,PHP 的後端和頁面,另外藉助 Apeche Cordova 用 HTML 實現了安卓的客戶端以供離線使用。最近用 Django 重新寫了,藉助 UpUp 的離線頁面,我不用爲了離線使用而再去生成一個安卓應用。

一般的網站,要使用 UpUp 是非常簡單的,只需要在網站的頂層文件夾引入對應的庫即可,但是 Django 的靜態文件一般是放在 static 下的,要想在項目根目錄下提供它們,可以這樣設置。

#項目的 url.py 文件
from django.urls import path
...
from . import views

urlpatterns = [
    ...
    path('upup.min.js', cache_page(60 * 60 * 48)(TemplateView.as_view(template_name="site/upup.min.js", 
  content_type='application/javascript', )), name='upup.min.js'),
    path('upup.sw.min.js', cache_page(60 * 60 * 48)(TemplateView.as_view(template_name="site/upup.sw.min.js", 
  content_type='application/javascript', )), name='upup.sw.min.js'),
]

然後將 upup.min.js 和 upup.sw.min.js 放在項目的模板目錄裏,比如 mysite/site/templates/site/ 中。

最後在 HTML 頁面中引用並設置需要離線的文件即可。

...
<script type="text/javascript" src="{% url 'upup.min.js' %}"></script>
<script type="text/javascript" src="{% static 'site/js/jquery-3.6.0.min.js' %}"></script>
<script>
UpUp.start({
  'content-url': '/site/',
  'assets': ['/static/site/js/jquery-3.6.0.min.js',
   '/static/site/js/site.js', '/static/site/css/site.css']

});
</script>

離線站點調試的過程中遇到的另外一個問題是,離線站點必須是 HTTPS 類型的加密頁面,但是本地配置 HTTPS 又略嫌繁瑣,其實只要再 Chrome 裏添加例外(chrome://flags/#enable-site-per-process)即可。火狐我還不知道要怎麼添加,最近安卓火狐的一系列更新都不盡人意,實在是有些令人擔憂。

分類
软件

心臟異常檢測 Heart Anomaly Detection

藉助手機的麥克風來錄製心跳,然後使用 MananAgarwal 訓練的 Heartbeat-Classifier 模型來判斷心跳是否異常。經過測試,我和 Emanon 的心跳都是正常。我也有嘗試使用其他項目的實驗數據來測試這個模型,對於「二尖瓣迴流」和「主動脈狹窄」的異常心跳都成功的識別了出來。不過它畢竟只是一個人工智能模型,如果有不舒服,還是應該儘早去看醫生。

安裝 Heart Anomaly Detection

這是一個 python3 的 tensorflow 項目。項目本身並不大(9 MB)但依賴大概有 600MB ,所以推薦使用虛擬環境來安裝

#安裝依賴
pip install tensorflow keras librosa 
#下載項目
git clone https://github.com/MananAgarwal/Heartbeat-Classifier.git
#測試運行
cd Heartbeat-Classifier
python testing.py heartbeat-to-classify.wav
#稍等片刻應該會成功輸出
Normal heartbeat
confidence: 0.9638266

使用手機錄製心跳

由於我不是蘋果手機用戶,所以無法推薦和測試適合的錄音應用。但是該模型的訓練數據中有用到蘋果手機(用iStethoscope Pro)錄製的心跳音頻,所以蘋果手機應該也是沒問題的。我的安卓手機使用的是 Audio Recorder,一個開源免費又小巧的錄音軟件(5.6 MB)。設置裏只需要把編碼格式改爲「.wav」,其他的保持不變即可滿足我們的需要。

找到手機的麥克風位置,一般都在手機的底部。找一個相對安靜的地方,點擊開始錄音(爲了避免誤觸屏幕,可以按一下電源鍵把屏幕關閉),把手機的麥克風的位置貼近心臟的位置,按住15秒左右,然後打開手機,停止錄音。我的建議是手機直接接觸皮膚來錄音,可能會獲得更好的音質。但是我給 Emanon 錄音的時候是隔着 T 恤的,倒也沒有影響的測試結果。

錄好的聲音文件默認保存在 /sdcard/Android/data/com.github.axet.audiorecorder/files/recordings 中,可以通過 USB 線或 Material Files 文件管理器將文件傳輸到電腦上。

最後使用上面的測試命令進行檢測,即可得到預測的心率狀態:Normal(正常)或 Abnormal(不正常),以及預測的信心值 confidence。

關於這個預測模型

該模型由博拉理工学院(印度皮拉尼)的 Manan Agarwal 和 Ankita Chakravarty 根據 The Classifying Heart Sounds Challenge 2011的數據集訓練而來。模型的訓練準確度爲 89.73,測試準確率爲 84.04。下面的文字摘自他們的論文說明:

我們的目標是提供一個可靠,快速且低成本的系統,讓未經培訓的一線衛生工作者或任何具有互聯網訪問權限的人都可以使用,以幫助確定是否應該將受試者推薦給專家診斷,尤其是在訪問臨床醫生和專家比較困難的地區。這也將有助於早期診斷心血管疾病,並大大降低這些死亡的潛在危險因素。